Azure-sdk-for-java: [BUG] NullPointerException when creating Spring-Cloud-Stream project with multiple Service-Bus binders and using namespace

[avpines]

Describe the bug
When creating a Spring-Cloud-Stream project with multiple Service-Bus binders (following the example in this repo), when using namespace instead of connection-string to the additional binder, a NullPointerException is thrown.

Exception or Stack Trace

org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'queueClientFactory' defined in com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusQueueAutoConfiguration: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.microsoft.azure.spring.integration.servicebus.factory.ServiceBusQueueClientFactory]: Factory method 'queueClientFactory' threw exception; nested exception is java.lang.NullPointerException
    at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:655)
    at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:635)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1336)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1176)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:556)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:516)
    at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:324)
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:226)
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:322)
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:202)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:897)
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:879)
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:551)
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:758)
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:750)
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:315)
    at org.springframework.boot.builder.SpringApplicationBuilder.run(SpringApplicationBuilder.java:140)
    at org.springframework.cloud.stream.binder.DefaultBinderFactory.getBinderInstance(DefaultBinderFactory.java:320)
    at org.springframework.cloud.stream.binder.DefaultBinderFactory.doGetBinder(DefaultBinderFactory.java:209)
    at org.springframework.cloud.stream.binder.DefaultBinderFactory.getBinder(DefaultBinderFactory.java:140)
    at org.springframework.cloud.stream.binding.BindingService.getBinder(BindingService.java:379)
    at org.springframework.cloud.stream.binding.BindingService.bindProducer(BindingService.java:268)
    at org.springframework.cloud.stream.function.StreamBridge.resolveDestination(StreamBridge.java:175)
    at org.springframework.cloud.stream.function.FunctionConfiguration$1.lambda$afterPropertiesSet$0(FunctionConfiguration.java:193)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.springframework.integration.handler.LambdaMessageProcessor.processMessage(LambdaMessageProcessor.java:97)
    at org.springframework.integration.router.AbstractMessageProcessingRouter.getChannelKeys(AbstractMessageProcessingRouter.java:84)
    at org.springframework.integration.router.AbstractMappingMessageRouter.determineTargetChannels(AbstractMappingMessageRouter.java:203)
    at org.springframework.integration.router.AbstractMessageRouter.handleMessageInternal(AbstractMessageRouter.java:177)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:62)
    at org.springframework.integration.handler.AbstractMessageHandler.onNext(AbstractMessageHandler.java:102)
    at org.springframework.integration.handler.AbstractMessageHandler.onNext(AbstractMessageHandler.java:37)
    at org.springframework.integration.endpoint.ReactiveStreamsConsumer.lambda$doStart$1(ReactiveStreamsConsumer.java:177)
    at reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160)
    at reactor.core.publisher.FluxDoFinally$DoFinallySubscriber.onNext(FluxDoFinally.java:123)
    at reactor.core.publisher.EmitterProcessor.drain(EmitterProcessor.java:426)
    at reactor.core.publisher.EmitterProcessor.onNext(EmitterProcessor.java:268)
    at reactor.core.publisher.FluxCreate$BufferAsyncSink.drain(FluxCreate.java:793)
    at reactor.core.publisher.FluxCreate$BufferAsyncSink.next(FluxCreate.java:718)
    at reactor.core.publisher.FluxCreate$SerializedSink.next(FluxCreate.java:153)
    at org.springframework.integration.channel.FluxMessageChannel.doSend(FluxMessageChannel.java:63)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:570)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:520)
    at org.springframework.integration.channel.FluxMessageChannel.lambda$subscribeTo$2(FluxMessageChannel.java:83)
    at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:189)
    at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.runAsync(FluxPublishOn.java:439)
    at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.run(FluxPublishOn.java:526)
    at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
    at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.microsoft.azure.spring.integration.servicebus.factory.ServiceBusQueueClientFactory]: Factory method 'queueClientFactory' threw exception; nested exception is java.lang.NullPointerException
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:185)
    at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:650)
    ... 57 common frames omitted
Caused by: java.lang.NullPointerException: null
    at com.microsoft.azure.spring.cloud.autoconfigure.servicebus.ServiceBusUtils.getNamespace(ServiceBusUtils.java:9)
    at com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusQueueAutoConfiguration.queueClientFactory(AzureServiceBusQueueAutoConfiguration.java:57)
    at com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusQueueAutoConfiguration$$EnhancerBySpringCGLIB$$b0f64529.CGLIB$queueClientFactory$0(<generated>)
    at com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusQueueAutoConfiguration$$EnhancerBySpringCGLIB$$b0f64529$$FastClassBySpringCGLIB$$6617ee6d.invoke(<generated>)
    at org.springframework.cglib.proxy.MethodProxy.invokeSuper(MethodProxy.java:244)
    at org.springframework.context.annotation.ConfigurationClassEnhancer$BeanMethodInterceptor.intercept(ConfigurationClassEnhancer.java:331)
    at com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusQueueAutoConfiguration$$EnhancerBySpringCGLIB$$b0f64529.queueClientFactory(<generated>)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:154)
    ... 58 common frames omitted

To Reproduce
Create a project with multiple binders that use namespace and not a connection string. Example application.yaml:

spring:
  cloud:
    azure:
      msi-enabled: true
      resource-group: my-resource-group
      region: us-west
      subscription-id: 11111111-1111-1111-1111-111111111111
      service-bus:
        namespace: my-namespace-1
    stream:
      function:
        definition: produce
      binders:
        queue-1-binder:
          type: servicebus-queue
          defaultCandidate: false
          environment:
            spring:
              cloud:
                azure:
                  msi-enabled: true
                  resource-group: my-resource-group
                  region: us-west
                  subscription-id: 11111111-1111-1111-1111-111111111111
                  service-bus:
                    namespace: my-namespace-2
      bindings:
        produce-out-0:
          destination: queue-a-0
        my-producer-output-1:
          destination: queue-a-1
          binder: queue-1-binder

Code Snippet
It seems that when creating the additional binder, in AzureServiceBusQueueAutoConfiguration.java, resourceManagerProvider is null, and therefore the else clause is triggered and the connection-string is checked:

 if (resourceManagerProvider != null) {
    clientFactory.setResourceManagerProvider(resourceManagerProvider);
    clientFactory.setNamespace(serviceBusProperties.getNamespace());
} else {
    TelemetryCollector.getInstance().addProperty(SERVICE_BUS_QUEUE, NAMESPACE,
        ServiceBusUtils.getNamespace(connectionString));
 }

Then in ServiceBusUtils there's an assumption that the connection-string cannot be null:

public static String getNamespace(String connectionString) {
    String prefix = "Endpoint=sb://";
    int start = connectionString.indexOf(prefix) + prefix.length();
    return connectionString.substring(start, connectionString.indexOf('.', start));
}

Expected behavior
The binder should connect just like when using a connection-string.

Setup (please complete the following information):

• OS: Tested on AKS
• Version of the Library used: 1.2.7, 1.2.8

Information Checklist
Kindly make sure that you have added all the following information above and checkoff the required fields otherwise we will treat the issuer as an incomplete report

• [x] Bug Description Added
• [x] Repro Steps Added
• [x] Setup information Added

[avpines]

This is a blocker for us for going to production, we must use MSI and namespace or it won't pass our security and compliance review.

[joshfree]

Thanks for reporting this issue @avpines. @hemanttanwar from the Azure SDK team will follow up shortly.

[hemanttanwar]

@avpines We are looking into issue and will provide you update soon. Appreciate you providing us details here.

[saragluna]

@avpines This was caused by a bug in our code and we're trying to fix it. So I'd like to double-check with you that provisioning the service buses via the code is not one of your intentions and you guys want to connect to the service buses using MSI?

[avpines]

@saragluna - Yep, we want to connect to multiple service buses via MSI. Thanks!

[saragluna]

Hi @avpines, fix #15755 has been merged to the master branch. The repo will publish daily built packages to the dev feed and please follow this to get the daily built packages. Please let us know whether this solves your problem.

[avpines]

Thanks @saragluna!

It seems that the last build com.microsoft.azure:spring-cloud-azure-servicebus-queue-stream-binder 1.3.0-alpha.20200925.0 was built on Saturday (prior to the merge). Do you know when another build is scheduled? (It doesn't seem to be daily for all artifacts).

[saragluna]

@avpines sorry for that. I checked the daily build pipeline and it kept failing because of some reason. I will try to re-run it and let you know once there's an available package.

[avpines]

@saragluna I see that the last build is still on Saturday. Any updates or ETA when the pipeline will be fixed? Thanks!

[saragluna]

@avpines sorry for the late response, it was Chinese National Day holiday. We've just figured out what was wrong with our nightly built pipeline and it shall be fixed soon. I'll update you once it's done.

[saragluna]

@avpines our nightly build pipeline's been fixed. Could you please try it to see whether the issue has been solved? https://github.com/Azure/azure-sdk-for-java/blob/master/CONTRIBUTING.md#dev-feed

[avpines]

@saragluna it seems like the issue was not fixed, and now the connection string doesn't work either. I checked for com.azure:azure-spring-cloud-servicebus-queue-stream-binder:2.0.0-alpha.20201014.0 and
for both namespace and connection-string I am getting the following exception:

2020-10-15 15:37:36.367 ERROR 1 --- [   scheduling-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'unknown.channel.name'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[237], headers={id=70ad016d-5b75-4a5f-2387-f6a8a9db40f4, spring.cloud.stream.sendto.destination=producer-output-second, contentType=application/json, accept=application/json, timestamp=1602776256366}], failedMessage=GenericMessage [payload=byte[237], headers={id=70ad016d-5b75-4a5f-2387-f6a8a9db40f4, spring.cloud.stream.sendto.destination=producer-output-second, contentType=application/json, accept=application/json, timestamp=1602776256366}]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:76)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:578)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:520)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
    at org.springframework.integration.router.AbstractMessageRouter.doSend(AbstractMessageRouter.java:213)
    at org.springframework.integration.router.AbstractMessageRouter.handleMessageInternal(AbstractMessageRouter.java:195)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:62)
    at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:570)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:520)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
    at org.springframework.integration.endpoint.SourcePollingChannelAdapter.handleMessage(SourcePollingChannelAdapter.java:196)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.messageReceived(AbstractPollingEndpoint.java:444)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:428)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:376)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$null$3(AbstractPollingEndpoint.java:323)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
    at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$4(AbstractPollingEndpoint.java:320)
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
    at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:93)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[237], headers={id=70ad016d-5b75-4a5f-2387-f6a8a9db40f4, spring.cloud.stream.sendto.destination=producer-output-second, contentType=application/json, accept=application/json, timestamp=1602776256366}]
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:139)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72)
    ... 36 more

Comparing to the build _before_ the fix com.microsoft.azure:spring-cloud-azure-servicebus-queue-stream-binder:1.3.0-alpha.20200925.0, which works with connection-string and for namespace I am getting:

Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [com.microsoft.azure.spring.integration.servicebus.factory.ServiceBusQueueClientFactory]: Factory method 'queueClientFactory' threw exception; nested exception is java.lang.NullPointerException
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:185) ~[spring-beans-5.2.9.RELEASE.jar:5.2.9.RELEASE]
    at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:650) ~[spring-beans-5.2.9.RELEASE.jar:5.2.9.RELEASE]
    ... 60 common frames omitted
Caused by: java.lang.NullPointerException: null
    at com.microsoft.azure.spring.cloud.autoconfigure.servicebus.ServiceBusUtils.getNamespace(ServiceBusUtils.java:9) ~[spring-cloud-azure-autoconfigure-1.3.0-alpha.20200925.0.jar:na]
    at com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusQueueAutoConfiguration.queueClientFactory(AzureServiceBusQueueAutoConfiguration.java:57) ~[spring-cloud-azure-autoconfigure-1.3.0-alpha.20200925.0.jar:na]
    at com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusQueueAutoConfiguration$$EnhancerBySpringCGLIB$$a83b5eea.CGLIB$queueClientFactory$0(<generated>) ~[spring-cloud-azure-autoconfigure-1.3.0-alpha.20200925.0.jar:na]
    at com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusQueueAutoConfiguration$$EnhancerBySpringCGLIB$$a83b5eea$$FastClassBySpringCGLIB$$50ceacf9.invoke(<generated>) ~[spring-cloud-azure-autoconfigure-1.3.0-alpha.20200925.0.jar:na]
    at org.springframework.cglib.proxy.MethodProxy.invokeSuper(MethodProxy.java:244) ~[spring-core-5.2.9.RELEASE.jar:5.2.9.RELEASE]
    at org.springframework.context.annotation.ConfigurationClassEnhancer$BeanMethodInterceptor.intercept(ConfigurationClassEnhancer.java:331) ~[spring-context-5.2.9.RELEASE.jar:5.2.9.RELEASE]
    at com.microsoft.azure.spring.cloud.autoconfigure.servicebus.AzureServiceBusQueueAutoConfiguration$$EnhancerBySpringCGLIB$$a83b5eea.queueClientFactory(<generated>) ~[spring-cloud-azure-autoconfigure-1.3.0-alpha.20200925.0.jar:na]
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:na]
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
    at java.base/java.lang.reflect.Method.invoke(Method.java:566) ~[na:na]
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:154) ~[spring-beans-5.2.9.RELEASE.jar:5.2.9.RELEASE]
    ... 61 common frames omitted

Also, and I am not sure if it is related, I am getting an exception when running the same tests on the Event-Hubs binder for namespace: Caused by: java.lang.IllegalArgumentException: Event hubs connection string cannot be empty. I can elaborate more here or open a different issue for it, whichever you prefer.

[saragluna]

@avpines sorry for the binder not working as expected... I will look into it today.

[saragluna]

Hi @avpines, we've tried the com.azure:azure-spring-cloud-servicebus-queue-stream-binder:2.0.0-alpha.20201014.0 in VM, with the Managed Identity enabled, but cannot reproduce the error you encountered. Could you please share your sample project so we could try that?

[saragluna]

[avpines]

[ramya-rao-a]

[saragluna]

[saragluna]

[saragluna]