[BUG] Use Azure Resource Manager to retrieve Connection String doesn't work

[TechPreacher]

Describe the bug
When using the Spring Cloud Azure Stream Binder for Service Bus to receive a message from a Service Bus queue using a Service Principal that is in the "Contributor" role in the Service Bus but does NOT have the Azure Service Bus Data Sender or Azure Service Bus Data Receiver roles set, the Azure Resource Manager should be able to retrieve the Connection String using the Service Principal and thus be able to send or receive messages as outlined here
This however fails in 4.0.0-beta3 and the Service Principal is lacking the permission to listen on or send to a queue without the explicit Data Sender and Data Receiver roles.

Exception or Stack Trace

com.azure.messaging.servicebus.ServiceBusException: Unauthorized access. 'Listen' claim(s) are required to perform this operation. Resource: 'sb://simhub-servicebus-temp.servicebus.windows.net/create-replica'. TrackingId:2d762e1b035249449aadb9de4fcdf8b4_G25, SystemTracker:gateway7, Timestamp:2022-03-23T09:06:34, errorContext[NAMESPACE: simhub-servicebus-temp.servicebus.windows.net. ERROR CONTEXT: N/A, PATH: create-replica, REFERENCE_ID: create-replica_4f6648_1648026395300, LINK_CREDIT: 1]
	at com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient.mapError(ServiceBusReceiverAsyncClient.java:1496) ~[azure-messaging-servicebus-7.5.1.jar:7.5.1]
	at com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient.lambda$receiveMessagesWithContext$18(ServiceBusReceiverAsyncClient.java:799) ~[azure-messaging-servicebus-7.5.1.jar:7.5.1]
	at reactor.core.publisher.Flux.lambda$onErrorMap$28(Flux.java:6910) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onError(FluxOnErrorResume.java:94) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.doError(FluxPublishOn.java:511) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.checkTerminated(FluxPublishOn.java:549) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.runAsync(FluxPublishOn.java:432) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.run(FluxPublishOn.java:527) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.scheduler.ImmediateScheduler$ImmediateSchedulerWorker.schedule(ImmediateScheduler.java:84) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.trySchedule(FluxPublishOn.java:312) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.onError(FluxPublishOn.java:248) ~[reactor-core-3.4.11.jar:3.4.11]
	at com.azure.messaging.servicebus.FluxAutoLockRenew$LockRenewSubscriber.hookOnError(FluxAutoLockRenew.java:121) ~[azure-messaging-servicebus-7.5.1.jar:7.5.1]
	at reactor.core.publisher.BaseSubscriber.onError(BaseSubscriber.java:180) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxMap$MapSubscriber.onError(FluxMap.java:132) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxMap$MapSubscriber.onError(FluxMap.java:132) ~[reactor-core-3.4.11.jar:3.4.11]
	at com.azure.messaging.servicebus.implementation.ServiceBusReceiveLinkProcessor.onError(ServiceBusReceiveLinkProcessor.java:332) ~[azure-messaging-servicebus-7.5.1.jar:7.5.1]
	at com.azure.messaging.servicebus.implementation.ServiceBusReceiveLinkProcessor.lambda$onNext$4(ServiceBusReceiveLinkProcessor.java:221) ~[azure-messaging-servicebus-7.5.1.jar:7.5.1]
	at reactor.core.publisher.LambdaSubscriber.onError(LambdaSubscriber.java:149) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onError(FluxSubscribeOn.java:157) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxDistinct$DistinctFuseableSubscriber.onError(FluxDistinct.java:490) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replayNormal(FluxReplay.java:844) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replay(FluxReplay.java:944) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxReplay$ReplaySubscriber.onError(FluxReplay.java:1339) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxPeek$PeekSubscriber.onError(FluxPeek.java:222) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxMap$MapSubscriber.onError(FluxMap.java:132) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxDistinctUntilChanged$DistinctUntilChangedSubscriber.onError(FluxDistinctUntilChanged.java:162) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replayNormal(FluxReplay.java:844) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replay(FluxReplay.java:944) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.ReplayProcessor.tryEmitError(ReplayProcessor.java:488) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.SinkManySerialized.tryEmitError(SinkManySerialized.java:82) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.publisher.InternalManySink.emitError(InternalManySink.java:98) ~[reactor-core-3.4.11.jar:3.4.11]
	at com.azure.core.amqp.implementation.handler.Handler.onError(Handler.java:105) ~[azure-core-amqp-2.3.5.jar:2.3.5]
	at com.azure.core.amqp.implementation.handler.LinkHandler.handleRemoteLinkClosed(LinkHandler.java:113) ~[azure-core-amqp-2.3.5.jar:2.3.5]
	at com.azure.core.amqp.implementation.handler.LinkHandler.onLinkRemoteClose(LinkHandler.java:61) ~[azure-core-amqp-2.3.5.jar:2.3.5]
	at com.azure.core.amqp.implementation.handler.ReceiveLinkHandler.onLinkRemoteClose(ReceiveLinkHandler.java:193) ~[azure-core-amqp-2.3.5.jar:2.3.5]
	at org.apache.qpid.proton.engine.BaseHandler.handle(BaseHandler.java:176) ~[proton-j-0.33.8.jar:na]
	at org.apache.qpid.proton.engine.impl.EventImpl.dispatch(EventImpl.java:108) ~[proton-j-0.33.8.jar:na]
	at org.apache.qpid.proton.reactor.impl.ReactorImpl.dispatch(ReactorImpl.java:324) ~[proton-j-0.33.8.jar:na]
	at org.apache.qpid.proton.reactor.impl.ReactorImpl.process(ReactorImpl.java:291) ~[proton-j-0.33.8.jar:na]
	at com.azure.core.amqp.implementation.ReactorExecutor.run(ReactorExecutor.java:92) ~[azure-core-amqp-2.3.5.jar:2.3.5]
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68) ~[reactor-core-3.4.11.jar:3.4.11]
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28) ~[reactor-core-3.4.11.jar:3.4.11]
	at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) ~[na:na]
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) ~[na:na]
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) ~[na:na]
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na]
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na]
	at java.base/java.lang.Thread.run(Thread.java:829) ~[na:na]
Caused by: com.azure.core.amqp.exception.AmqpException: Unauthorized access. 'Listen' claim(s) are required to perform this operation. Resource: 'sb://simhub-servicebus-temp.servicebus.windows.net/create-replica'. TrackingId:2d762e1b035249449aadb9de4fcdf8b4_G25, SystemTracker:gateway7, Timestamp:2022-03-23T09:06:34, errorContext[NAMESPACE: simhub-servicebus-temp.servicebus.windows.net. ERROR CONTEXT: N/A, PATH: create-replica, REFERENCE_ID: create-replica_4f6648_1648026395300, LINK_CREDIT: 1]
	at com.azure.core.amqp.implementation.ExceptionUtil.toException(ExceptionUtil.java:85) ~[azure-core-amqp-2.3.5.jar:2.3.5]
	at com.azure.core.amqp.implementation.handler.LinkHandler.handleRemoteLinkClosed(LinkHandler.java:110) ~[azure-core-amqp-2.3.5.jar:2.3.5]
	... 15 common frames omitted

To Reproduce
Steps to reproduce the behavior:
Running the sample as outlined here fails.

Code Snippet
Default, auto-wired code:

    @Bean
    public Function<Message<T>, Message<T>> consume() {
        return message -> handler.handle(message);
    }

Expected behavior
The Service Principal should be able to connect to the Queue or Topic by using the Azure Resource Manager to retrieve a Connection String instead of requiring the Azure Service Bus Data Sender and Azure Service Bus Data Receiver roles.

Screenshots
n/a

Setup (please complete the following information):

• OS: Windows 11
• IDE: IntelliJ IDEA
• Library/Libraries: spring-cloud-azure-starter-stream-servicebus:4.0.0-beta3.
• Java version: Temurin 11: jdk-11.0.14.101-hotspot
• App Server/Environment: Pod running in AKS
• Frameworks: SpringBoot 2.5.6

Additional context
This issue is for documentation purpose. It has been addressed and fixed in 4.0.0-beta4

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

• Bug Description Added
• Repro Steps Added
• Setup information Added

[TechPreacher]

@yiliuTo : This is the but you fixed in 4.0.0-beta4. Thanks for that! Can I ask you to resolve this bug?

[yiliuTo]

@TechPreacher thanks for tracking this! Resolved by #27626

[ghost]

Hi @TechPreacher. Thank you for opening this issue and giving us the opportunity to assist. We believe that this has been addressed. If you feel that further discussion is needed, please add a comment with the text “/unresolve” to remove the “issue-addressed” label and continue the conversation.

[Netyyyy]

Here is a sample to use ARM to retrieve connection string

[yiliuTo]

Close for the issue has been resolved.

[TechPreacher]

Thanks again for the great help with this, @yiliuTo!