[BUG] Service Bus ServiceBusReceiverClient stop receiving message in a new thread

[liukun-msft]

Describe the bug
When use ServiceBusReceiverClient.receiveMessages(int maxMessages) to synchronous receive message on thread-1, kill this thread in the middle of iterating the received messages. Then bring up a new thread-2 to receive messages, the receiver client doesn't retrieve any messages.

To Reproduce

Repro is not stable, only happened sometimes:

1. Enable debug mode
2. Empty the service bus queue
3. Start receiver and wait until logs contains one message "Current work updated."
4. Start sender to send messages
5. Throw an exception in receiver thread and bring up a new thread to keep receiving

Code Snippet

Repro test code:
https://github.com/conniey/servicebus-demo/blob/main/src/main/java/org/example/SyncReceiveThrows.java

Exception or Stack Trace

13:56:47.118 [reactor-executor-1] DEBUG c.a.c.a.i.handler.ReceiveLinkHandler - {"az.sdk.message":"onDelivery.","connectionId":"MF_5577fe_1646718906637","errorCondition":null,"errorDescription":null,"entityPath":"testqueue","linkName":"testqueue_7552d7_1646718906715","updatedLinkCredit":6,"remoteCredit":6,"delivery.isPartial":false,"delivery.isSettled":false}
13:56:47.118 [reactor-executor-1] DEBUG c.a.m.s.i.ServiceBusReactorReceiver - entityPath[testqueue], linkName[testqueue_7552d7_1646718906715], deliveryTag[60a8a670-214b-4f8c-a1eb-06db47d67944], state[Accepted{}] Received update disposition delivery.
13:56:47.118 [reactor-executor-1] DEBUG c.a.m.s.ServiceBusReceiverAsyncClient - testqueue: Update completed. Disposition: COMPLETED. Lock: 60a8a670-214b-4f8c-a1eb-06db47d67944.
13:56:47.124 [reactor-executor-1] DEBUG c.a.c.a.i.handler.ReceiveLinkHandler - {"az.sdk.message":"onDelivery.","connectionId":"MF_5577fe_1646718906637","errorCondition":null,"errorDescription":null,"entityPath":"testqueue","linkName":"testqueue_7552d7_1646718906715","updatedLinkCredit":6,"remoteCredit":5,"delivery.isPartial":false,"delivery.isSettled":false}
round[1] #[15] messageId[2] 
---- CREATING 2 ----
13:56:47.124 [Thread-3] DEBUG c.a.m.s.SynchronousMessageSubscriber - workId[2] numberOfEvents[10] timeout[PT4M5.6S] First work in queue. Requesting upstream if needed.
Exception in thread "Thread-2" java.lang.IllegalStateException: Test error occurs. index: 15
	at com.azure.servicebus.scenarios.SyncReceiveThrows.lambda$createNewThread$0(SyncReceiveThrows.java:53)
	at java.base/java.lang.Thread.run(Thread.java:834)
13:56:47.551 [reactor-executor-1] DEBUG c.a.c.a.i.handler.ReceiveLinkHandler - {"az.sdk.message":"onDelivery.","connectionId":"MF_5577fe_1646718906637","errorCondition":null,"errorDescription":null,"entityPath":"testqueue","linkName":"testqueue_7552d7_1646718906715","updatedLinkCredit":5,"remoteCredit":4,"delivery.isPartial":false,"delivery.isSettled":false}
13:56:47.978 [reactor-executor-1] DEBUG c.a.c.a.i.handler.ReceiveLinkHandler - {"az.sdk.message":"onDelivery.","connectionId":"MF_5577fe_1646718906637","errorCondition":null,"errorDescription":null,"entityPath":"testqueue","linkName":"testqueue_7552d7_1646718906715","updatedLinkCredit":4,"remoteCredit":3,"delivery.isPartial":false,"delivery.isSettled":false}
13:56:48.398 [reactor-executor-1] DEBUG c.a.c.a.i.handler.ReceiveLinkHandler - {"az.sdk.message":"onDelivery.","connectionId":"MF_5577fe_1646718906637","errorCondition":null,"errorDescription":null,"entityPath":"testqueue","linkName":"testqueue_7552d7_1646718906715","updatedLinkCredit":3,"remoteCredit":2,"delivery.isPartial":false,"delivery.isSettled":false}
13:56:48.819 [reactor-executor-1] DEBUG c.a.c.a.i.handler.ReceiveLinkHandler - {"az.sdk.message":"onDelivery.","connectionId":"MF_5577fe_1646718906637","errorCondition":null,"errorDescription":null,"entityPath":"testqueue","linkName":"testqueue_7552d7_1646718906715","updatedLinkCredit":2,"remoteCredit":1,"delivery.isPartial":false,"delivery.isSettled":false}
13:56:49.245 [reactor-executor-1] DEBUG c.a.c.a.i.handler.ReceiveLinkHandler - {"az.sdk.message":"onDelivery.","connectionId":"MF_5577fe_1646718906637","errorCondition":null,"errorDescription":null,"entityPath":"testqueue","linkName":"testqueue_7552d7_1646718906715","updatedLinkCredit":1,"remoteCredit":0,"delivery.isPartial":false,"delivery.isSettled":false}
13:56:49.245 [reactor-executor-1] DEBUG c.a.c.a.i.ReactorReceiver - {"az.sdk.message":"There are no credits to add.","connectionId":"MF_5577fe_1646718906637","linkName":"testqueue_7552d7_1646718906715","credits":"0"}
//Stuck here

Note that logs doesn't have message "Current work updated" for work[2] here, which means work[2] haven't been updated to execute.

Expected behavior

Receiver should keep receiving message in the new thread.

Setup (please complete the following information):

• Library/Libraries: com.azure:azure-messaging-servicebus:7.5.1
• Java version: 11

Additional context
This issue are not in azure-messaging-servicebus:7.4.2

[liukun-msft]

Reason

ServiceBusReceiverClient will wrap a receiving task as SynchronousReceiveWork and put it into SynchronousMessageSubscriber.
Let's say we start Thread-1 to receive message, it will wrap Work 1, and the work is scheduled to run on another thread. So if we kill thread-1 in the middle of processing Work 1 and bring up a new Thread-2 to keep receiving messages, base on current logic, it won't update the new coming Work 2 but just put Work 2 into the work queue. Seems there is no logic to process rest of works in the queue, so the receiver will not receive any messages and get stuck.

Mitigate

Sleep a while before bring up a new thread to wait previous work finished.

Fix

Queue the work when current work is not terminated and then update the work when previous work is completed.

[liukun-msft]

Fix (updated)

Test dev build (only for test and should not use in any prod env):

1. Download and unzip below devbuild.zip:
   0316-devbuild.zip

2. Import azure-core-amqp-2.5.0-beta.1.jar and azure-messaging-servicebus-7.7.0-beta.2.jar to your project.

3. Make sure below dependencies are also imported and version are same.

<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-core</artifactId>
    <version>1.26.0</version> 
</dependency>
<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-core-http-netty</artifactId>
    <version>1.11.8</version>
</dependency>
<dependency>
    <groupId>com.microsoft.azure</groupId>
    <artifactId>qpid-proton-j-extensions</artifactId>
    <version>1.2.4</version>
</dependency>
<dependency>
    <groupId>org.apache.qpid</groupId>
    <artifactId>proton-j</artifactId>
    <version>0.33.8</version> 
</dependency>

ARCHIEVE version: 0310-devbuild.zip
This version has issue when receive message = 1. Because when receive work is completed, it update next work first and then minus 1 for the REQUESTED. So when next receive work request upstream, the client thought it already request 1, which actually belongs to the completed work.