Avoid error in one faulty receiver bypassing to parent connection hosting other healthy receivers, ensure ExceutionRejectedException is handled in all places that uses Dispatcher.invoke (Azure#27839)

Author: anuchandy

eng/versioning/version_client.txt

@@ -354,6 +354,7 @@ com.azure.tools:azure-sdk-build-tool;1.0.0-beta.1;1.0.0-beta.2
 unreleased_com.azure:azure-aot-graalvm-support;1.0.0-beta.1
 unreleased_com.azure:azure-aot-graalvm-support-netty;1.0.0-beta.1
 unreleased_com.azure:azure-aot-graalvm-perf;1.0.0-beta.1
+unreleased_com.azure:azure-core-amqp;2.5.0-beta.1
 unreleased_com.azure:azure-core;1.27.0-beta.1
 
 # Released Beta dependencies: Copy the entry from above, prepend "beta_", remove the current

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorReceiver.java

@@ -83,6 +83,11 @@ protected ReactorReceiver(AmqpConnection amqpConnection, String entityPath, Rece
                 return Mono.create(sink -> {
                     try {
                         this.dispatcher.invoke(() -> {
+                            if (isDisposed()) {
+                                sink.error(new IllegalStateException(
+                                    "Cannot decode delivery when ReactorReceiver instance is closed."));
+                                return;
+                            }
                             final Message message = decodeDelivery(delivery);
                             final int creditsLeft = receiver.getRemoteCredit();
 

sdk/eventhubs/azure-messaging-eventhubs/pom.xml

@@ -42,7 +42,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core-amqp</artifactId>
-      <version>2.4.1</version> <!-- {x-version-update;com.azure:azure-core-amqp;dependency} -->
+      <version>2.5.0-beta.1</version> <!-- {x-version-update;unreleased_com.azure:azure-core-amqp;dependency} -->
     </dependency>
 
     <!-- Test dependencies -->

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/AmqpReceiveLinkProcessor.java

@@ -283,10 +283,20 @@ public void onNext(AmqpReceiveLink next) {
                     }),
                 next.receive()
                     .onBackpressureBuffer(maxQueueSize, BufferOverflowStrategy.ERROR)
-                    .subscribe(message -> {
-                        messageQueue.add(message);
-                        drain();
-                    }));
+                    .subscribe(
+                        message -> {
+                            messageQueue.add(message);
+                            drain();
+                        },
+                        error -> {
+                            // When the receive on AmqpReceiveLink (e.g., ReactorReceiver) terminates
+                            // with an error, we expect the recovery to happen in response to the terminal events
+                            // in link EndpointState Flux.
+                            logger.atVerbose()
+                                .addKeyValue(LINK_NAME_KEY, linkName)
+                                .addKeyValue(ENTITY_PATH_KEY, entityPath)
+                                .log("Receiver is terminated.", error);
+                        }));
         }
 
         disposeReceiver(oldChannel);

sdk/servicebus/azure-messaging-servicebus/pom.xml

@@ -53,7 +53,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core-amqp</artifactId>
-      <version>2.4.1</version> <!-- {x-version-update;com.azure:azure-core-amqp;dependency} -->
+      <version>2.5.0-beta.1</version> <!-- {x-version-update;unreleased_com.azure:azure-core-amqp;dependency} -->
     </dependency>
     <dependency>
       <groupId>com.azure</groupId>

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusReactorReceiver.java

@@ -49,6 +49,7 @@
 import java.util.StringJoiner;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -265,7 +266,7 @@ private Mono<Void> updateDispositionInternal(String lockToken, DeliveryState del
                     unsettled.disposition(deliveryState);
                     pendingUpdates.put(lockToken, workItem);
                 });
-            } catch (IOException error) {
+            } catch (IOException | RejectedExecutionException error) {
                 sink.error(new AmqpException(false, "updateDisposition failed while dispatching to Reactor.",
                     error, handler.getErrorContext(receiver)));
             }
@@ -348,7 +349,7 @@ private void updateOutcome(String lockToken, Delivery delivery) {
                     workItem.resetStartTime();
                     try {
                         provider.getReactorDispatcher().invoke(() -> delivery.disposition(workItem.getDeliveryState()));
-                    } catch (IOException error) {
+                    } catch (IOException | RejectedExecutionException error) {
                         final Throwable amqpException = logger.atError()
                             .addKeyValue(LOCK_TOKEN_KEY, lockToken)
                             .log(new AmqpException(false,

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusReceiveLinkProcessor.java

@@ -217,14 +217,24 @@ public void onNext(ServiceBusReceiveLink next) {
             next.setEmptyCreditListener(() -> 0);
 
             currentLinkSubscriptions = Disposables.composite(
-                next.receive().publishOn(Schedulers.boundedElastic()).subscribe(message -> {
-                    synchronized (queueLock) {
-                        messageQueue.add(message);
-                        pendingMessages.incrementAndGet();
-                    }
+                next.receive().publishOn(Schedulers.boundedElastic()).subscribe(
+                    message -> {
+                        synchronized (queueLock) {
+                            messageQueue.add(message);
+                            pendingMessages.incrementAndGet();
+                        }
 
-                    drain();
-                }),
+                        drain();
+                    },
+                    error -> {
+                        // When the receive on AmqpReceiveLink (e.g., ServiceBusReactorReceiver) terminates
+                        // with an error, we expect the recovery to happen in response to the terminal events
+                        // in link EndpointState Flux.
+                        logger.atVerbose()
+                            .addKeyValue(LINK_NAME_KEY, linkName)
+                            .addKeyValue(ENTITY_PATH_KEY, entityPath)
+                            .log("Receiver is terminated.", error);
+                    }),
                 next.getEndpointStates().subscribeOn(Schedulers.boundedElastic()).subscribe(
                     state -> {
                         // Connection was successfully opened, we can reset the retry interval.