Add timeout to cleaning up sender/receiver links. (Azure#23381)

* Add logging for Handler and do not fail fast when complete cannot be emitted on close.
Updating tests.

* Add documentation to LinkHandler. Add null check. Explicitly call super.close().

* Do not call super.close() when close is called. We set the endpoint state to close, but leave the base LinkHandler to complete getEndpointStates().
Adding documentation.

* Adding tests for LinkHandler.

* Being explicit about calling handler.super(). Doing NPE checks.

* Updating log messages to say how many pending sends we are disposing of.

* Add timeout with close when disposing of RequestResponseChannel.

* Adding verbose logs to ReactorConnection.

* Adding SendLinkHandler test and ReceiveLinkHandlerTest to take care of localOpen and closes.

* Fixing break in ReactorReceiverTest

* Adding to ReactorConnectionTest.

* Adding subscribeOn.

* Emitting complete endpoints signal.

* Adding tests for close.

* Add CHANGELOG entry.

Author: conniey

sdk/core/azure-core-amqp/CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Fixed a bug where SendTimeout-timer thread was not being disposed of resulting in lingering
   threads when a send link was remotely closed.
+- Fixed a bug where ReactorConnection waited indefinitely for CBS node to complete closing. The underlying problem is that the RequestResponseChannel's sender and receiver links were not active, so they would wait forever for a remote close signal.
 
 ## 2.3.0 (2021-07-01)
 

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

@@ -431,23 +431,33 @@ Mono<Void> closeAsync(AmqpShutdownSignal shutdownSignal) {
             Flux.fromStream(managementNodes.values().stream()).flatMap(node -> node.closeAsync()));
 
         final Mono<Void> closeReactor = Mono.fromRunnable(() -> {
+            logger.verbose("connectionId[{}] Scheduling closeConnection work.", connectionId);
             final ReactorDispatcher dispatcher = reactorProvider.getReactorDispatcher();
 
-            try {
-                if (dispatcher != null) {
-                    dispatcher.invoke(this::closeConnectionWork);
-                } else {
+            if (dispatcher != null) {
+                try {
+                    dispatcher.invoke(() -> closeConnectionWork());
+                } catch (IOException | RejectedExecutionException e) {
+                    logger.warning("connectionId[{}] Error while scheduling closeConnection work. Manually disposing.",
+                        connectionId, e);
                     closeConnectionWork();
                 }
-            } catch (IOException | RejectedExecutionException e) {
-                logger.warning("connectionId[{}] Error while scheduling closeConnection work. Manually disposing.",
-                    connectionId, e);
+            } else {
                 closeConnectionWork();
             }
         });
 
-        return Mono.whenDelayError(cbsCloseOperation, managementNodeCloseOperations)
-            .then(closeReactor)
+        return Mono.whenDelayError(
+            cbsCloseOperation.doFinally(signalType ->
+                logger.verbose("connectionId[{}] signalType[{}] Closed CBS node.", connectionId, signalType)),
+            managementNodeCloseOperations.doFinally(signalType ->
+                logger.verbose("connectionId[{}] signalType[{}] Closed management nodes.", connectionId,
+                    signalType)))
+
+
+            .then(closeReactor.doFinally(signalType ->
+                logger.verbose("connectionId[{}] signalType[{}] Closed reactor dispatcher.", connectionId,
+                    signalType)))
             .then(isClosedMono.asMono());
     }
 
@@ -469,8 +479,9 @@ private synchronized void closeConnectionWork() {
         final ArrayList<Mono<Void>> closingSessions = new ArrayList<>();
         sessionMap.values().forEach(link -> closingSessions.add(link.isClosed()));
 
-        final Mono<Void> closedExecutor = executor != null ? Mono.defer(() ->  {
+        final Mono<Void> closedExecutor = executor != null ? Mono.defer(() -> {
             synchronized (this) {
+                logger.info("connectionId[{}] Closing executor.", connectionId);
                 return executor.closeAsync();
             }
         }) : Mono.empty();
@@ -535,7 +546,7 @@ private synchronized Connection getOrCreateConnection() throws IOException {
 
             // To avoid inconsistent synchronization of executor, we set this field with the closeAsync method.
             // It will not be kicked off until subscribed to.
-            final Mono<Void> executorCloseMono = Mono.defer(() ->  {
+            final Mono<Void> executorCloseMono = Mono.defer(() -> {
                 synchronized (this) {
                     return executor.closeAsync();
                 }

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

@@ -651,13 +651,13 @@ private void completeClose() {
      * @param error Error to pass to pending sends.
      */
     private void handleError(Throwable error) {
-        final String logMessage = isDisposed.getAndSet(true)
-            ? "This was already disposed. Dropping error."
-            : "Disposing pending sends with error.";
-        logger.verbose("connectionId[{}] entityPath[{}] linkName[{}] {}", handler.getConnectionId(), entityPath,
-            getLinkName(), logMessage, error);
-
         synchronized (pendingSendLock) {
+            final String logMessage = isDisposed.getAndSet(true)
+                ? "This was already disposed. Dropping error."
+                : String.format("Disposing of %s pending sends with error.", pendingSendsMap.size());
+            logger.verbose("connectionId[{}] entityPath[{}] linkName[{}] {}", handler.getConnectionId(), entityPath,
+                getLinkName(), logMessage, error);
+
             pendingSendsMap.forEach((key, value) -> value.error(error));
             pendingSendsMap.clear();
             pendingSendsQueue.clear();
@@ -667,17 +667,18 @@ private void handleError(Throwable error) {
     }
 
     private void handleClose() {
-        final String logMessage = isDisposed.getAndSet(true)
-            ? "This was already disposed."
-            : "Disposing pending sends.";
-        logger.verbose("connectionId[{}] entityPath[{}] linkName[{}] {}", handler.getConnectionId(), entityPath,
-            getLinkName(), logMessage);
-
         final String message = String.format("Could not complete sends because link '%s' for '%s' is closed.",
             getLinkName(), entityPath);
         final AmqpErrorContext context = handler.getErrorContext(sender);
 
         synchronized (pendingSendLock) {
+            final String logMessage = isDisposed.getAndSet(true)
+                ? "This was already disposed."
+                : String.format("Disposing of '%s' pending sends.", pendingSendsMap.size());
+
+            logger.verbose("connectionId[{}] entityPath[{}] linkName[{}] {}", handler.getConnectionId(), entityPath,
+                getLinkName(), logMessage);
+
             pendingSendsMap.forEach((key, value) -> value.error(new AmqpException(true, message, context)));
             pendingSendsMap.clear();
             pendingSendsQueue.clear();

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

@@ -40,6 +40,7 @@
 import java.util.UUID;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -170,6 +171,8 @@ protected RequestResponseChannel(AmqpConnection amqpConnection, String connectio
                 onTerminalState("SendLinkHandler");
             }),
 
+            //TODO (conniey): Do we need this if we already close the request response nodes when the
+            // connection.closeWork is executed? It would be preferred to get rid of this circular dependency.
             amqpConnection.getShutdownSignals().next().flatMap(signal -> {
                 logger.verbose("connectionId[{}] linkName[{}]: Shutdown signal received.", connectionId, linkName);
                 return closeAsync();
@@ -201,15 +204,33 @@ public Flux<AmqpEndpointState> getEndpointStates() {
 
     @Override
     public Mono<Void> closeAsync() {
+        final Mono<Void> closeOperationWithTimeout = closeMono.asMono()
+            .timeout(retryOptions.getTryTimeout())
+            .onErrorResume(TimeoutException.class, error -> {
+                return Mono.fromRunnable(() -> {
+                    logger.info("connectionId[{}] linkName[{}] Timed out waiting for RequestResponseChannel to complete"
+                            + "closing. Manually closing.",
+                        connectionId, linkName, error);
+
+                    onTerminalState("SendLinkHandler");
+                    onTerminalState("ReceiveLinkHandler");
+                });
+            })
+            .subscribeOn(Schedulers.boundedElastic());
+
         if (isDisposed.getAndSet(true)) {
-            return closeMono.asMono().subscribeOn(Schedulers.boundedElastic());
+            logger.verbose("connectionId[{}] linkName[{}] Channel already closed.", connectionId, linkName);
+            return closeOperationWithTimeout;
         }
 
         logger.verbose("connectionId[{}] linkName[{}] Closing request/response channel.", connectionId, linkName);
 
         return Mono.fromRunnable(() -> {
             try {
                 provider.getReactorDispatcher().invoke(() -> {
+                    logger.verbose("connectionId[{}] linkName[{}] Closing send link and receive link.",
+                        connectionId, linkName);
+
                     sendLink.close();
                     receiveLink.close();
                 });
@@ -219,7 +240,7 @@ public Mono<Void> closeAsync() {
                 sendLink.close();
                 receiveLink.close();
             }
-        }).publishOn(Schedulers.boundedElastic()).then(closeMono.asMono());
+        }).subscribeOn(Schedulers.boundedElastic()).then(closeOperationWithTimeout);
     }
 
     public boolean isDisposed() {
@@ -365,12 +386,21 @@ private void handleError(Throwable error, String message) {
     }
 
     private void onTerminalState(String handlerName) {
+        if (pendingDisposes.get() == 0) {
+            logger.verbose("connectionId[{}] linkName[{}]: Already disposed send/receive links.");
+            return;
+        }
+
         final int remaining = pendingDisposes.decrementAndGet();
         logger.verbose("connectionId[{}] linkName[{}]: {} disposed. Remaining: {}",
             connectionId, linkName, handlerName, remaining);
 
         if (remaining == 0) {
             subscriptions.dispose();
+
+            endpointStates.emitComplete(((signalType, emitResult) -> onEmitSinkFailure(signalType, emitResult,
+                "Could not emit complete signal.")));
+
             closeMono.emitEmpty((signalType, emitResult) -> onEmitSinkFailure(signalType, emitResult,
                 handlerName + ". Error closing mono."));
         }

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

@@ -34,12 +34,12 @@ public abstract class Handler extends BaseHandler implements Closeable {
      *     messages are brokered through an intermediary.
      * @param logger Logger to use for messages.
      *
-     * @throws NullPointerException if {@code connectionId} or {@code hostname} is null.
+     * @throws NullPointerException if {@code connectionId}, {@code hostname}, or {@code logger} is null.
      */
     Handler(final String connectionId, final String hostname, ClientLogger logger) {
         this.connectionId = Objects.requireNonNull(connectionId, "'connectionId' cannot be null.");
         this.hostname = Objects.requireNonNull(hostname, "'hostname' cannot be null.");
-        this.logger = logger;
+        this.logger = Objects.requireNonNull(logger, "'logger' cannot be null.");
     }
 
     /**
@@ -68,11 +68,14 @@ public String getHostname() {
      * @return The endpoint states of the handler.
      */
     public Flux<EndpointState> getEndpointStates() {
+        // In previous incarnations, we used .distinct(). It hashed all the previous values and would only push values
+        // that were not seen yet. What we want is only to push endpoint states that are unique from the previous one.
         return endpointStates.asFlux().distinctUntilChanged();
     }
 
     /**
-     * Emits the next endpoint. If the previous endpoint was emitted, it is skipped.
+     * Emits the next endpoint. If the previous endpoint was emitted, it is skipped. If the handler is closed, the
+     * endpoint state is not emitted.
      *
      * @param state The next endpoint state to emit.
      */
@@ -117,7 +120,14 @@ public void close() {
             return;
         }
 
-        endpointStates.emitNext(EndpointState.CLOSED, Sinks.EmitFailureHandler.FAIL_FAST);
+        // This is fine in the case that someone called onNext(EndpointState.CLOSED) and then called handler.close().
+        // We want to ensure that the next endpoint subscriber does not believe the handler is alive still.
+        endpointStates.emitNext(EndpointState.CLOSED, (signalType, emitResult) -> {
+            logger.info("connectionId[{}] signal[{}] result[{}] Could not emit closed endpoint state.", connectionId,
+                signalType, emitResult);
+
+            return false;
+        });
 
         endpointStates.emitComplete((signalType, emitResult) -> {
             logger.verbose("connectionId[{}] result[{}] Could not emit complete.", connectionId, emitResult);

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

@@ -12,15 +12,36 @@
 import org.apache.qpid.proton.engine.Event;
 import org.apache.qpid.proton.engine.Link;
 
+import java.util.Objects;
+
 import static com.azure.core.amqp.implementation.AmqpErrorCode.TRACKING_ID_PROPERTY;
 import static com.azure.core.amqp.implementation.ClientConstants.NOT_APPLICABLE;
 
+/**
+ * Base class for AMQP links.
+ *
+ * @see SendLinkHandler
+ * @see ReceiveLinkHandler
+ */
 abstract class LinkHandler extends Handler {
     private final String entityPath;
 
+    /**
+     * Creates an instance with the parameters.
+     *
+     * @param connectionId Identifier for the connection.
+     * @param hostname Hostname of the connection. This could be the DNS hostname or the IP address of the
+     *     connection. Usually of the form {@literal "<your-namespace>.service.windows.net"} but can change if the
+     *     messages are brokered through an intermediary.
+     * @param entityPath The address within the message broker for this link.
+     * @param logger Logger to use for messages.
+     *
+     * @throws NullPointerException if {@code connectionId}, {@code hostname}, {@code entityPath}, or {@code logger} is
+     * null.
+     */
     LinkHandler(String connectionId, String hostname, String entityPath, ClientLogger logger) {
         super(connectionId, hostname, logger);
-        this.entityPath = entityPath;
+        this.entityPath = Objects.requireNonNull(entityPath, "'entityPath' cannot be null.");
     }
 
     @Override
@@ -51,7 +72,10 @@ public void onLinkFinal(Event event) {
             ? event.getLink().getName()
             : NOT_APPLICABLE;
         logger.info("onLinkFinal connectionId[{}], linkName[{}]", getConnectionId(), linkName);
-        close();
+
+        // Be explicit about wanting to call Handler.close(). When we receive onLinkFinal, the service and proton-j are
+        // releasing this link. So we want to complete the endpoint states.
+        super.close();
     }
 
     public AmqpErrorContext getErrorContext(Link link) {
@@ -88,7 +112,7 @@ eventName, getConnectionId(), link.getName(),
 
             onError(exception);
         } else {
-            close();
+            super.close();
         }
     }
 }