Avoid concurrent recovery attempts

Author: acogoluegnes

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConnection.java

@@ -322,14 +322,15 @@ TopologyListener createTopologyListener(AmqpConnectionBuilder builder) {
             // nothing to do in this listener
             return;
           }
+          AmqpException exception = ExceptionUtils.convert(event.failureCause());
+          LOGGER.debug("Converted native exception to {}", exception.getClass().getSimpleName());
           if (this.recoveringConnection.get()) {
             LOGGER.debug(
                 "Filtering recovery task scheduling, connection recovery of '{}' already in progress",
                 this.name());
+            this.releaseManagementResources(exception);
             return;
           }
-          AmqpException exception = ExceptionUtils.convert(event.failureCause());
-          LOGGER.debug("Converted native exception to {}", exception.getClass().getSimpleName());
 
           if (RECOVERY_PREDICATE.test(exception) && this.state() != OPENING) {
             LOGGER.debug(
@@ -343,6 +344,8 @@ TopologyListener createTopologyListener(AmqpConnectionBuilder builder) {
                       if (!this.recoveringConnection.get()) {
                         recoverAfterConnectionFailure(
                             recoveryConfiguration, name, exception, resultReference);
+                      } else {
+                        this.releaseManagementResources(exception);
                       }
                     });
           } else {
@@ -361,11 +364,11 @@ TopologyListener createTopologyListener(AmqpConnectionBuilder builder) {
   private void recoverAfterConnectionFailure(
       AmqpConnectionBuilder.AmqpRecoveryConfiguration recoveryConfiguration,
       String connectionName,
-      Exception failureCause,
+      AmqpException failureCause,
       AtomicReference<BiConsumer<org.apache.qpid.protonj2.client.Connection, DisconnectionEvent>>
           disconnectedHandlerReference) {
     LOGGER.info(
-        "Connection '{}' to '{}' has been disconnected, trying to recover.",
+        "Connection '{}' to '{}' has been disconnected, initializing recovery.",
         this.name(),
         this.currentConnectionLabel());
     LOGGER.debug("Notifying listeners of connection '{}'.", this.name());
@@ -376,11 +379,11 @@ private void recoverAfterConnectionFailure(
     this.nativeSession = null;
     this.connectionAddress = null;
     LOGGER.debug("Releasing management resource of connection '{}'.", this.name());
-    this.releaseManagementResources();
+    this.releaseManagementResources(failureCause);
     CompletableFuture<NativeConnectionWrapper> ncwFuture;
     if (this.recoveringConnection.compareAndSet(false, true)) {
       this.recoveringConnection.set(true);
-      LOGGER.debug("Connection attempt for '{}'.", this.name());
+      LOGGER.debug("Scheduling connection attempt for '{}'.", this.name());
       ncwFuture =
           recoverNativeConnection(
               recoveryConfiguration, connectionName, disconnectedHandlerReference);
@@ -394,7 +397,6 @@ private void recoverAfterConnectionFailure(
             ncw -> {
               this.sync(ncw);
               LOGGER.debug("Reconnected '{}' to {}", this.name(), this.currentConnectionLabel());
-              this.recoveringConnection.set(false);
               try {
                 if (recoveryConfiguration.topology()) {
                   this.management.init();
@@ -407,13 +409,15 @@ private void recoverAfterConnectionFailure(
                 LOGGER.info(
                     "Recovered connection '{}' to {}", this.name(), this.currentConnectionLabel());
                 this.state(OPEN);
+                this.recoveringConnection.set(false);
               } catch (Exception ex) {
                 // likely InterruptedException or IO exception
                 LOGGER.warn(
                     "Error while trying to recover topology for connection '{}': {}",
                     this.name(),
                     ex.getMessage());
-                if (RECOVERY_PREDICATE.test(ex)) {
+                AmqpException amqpException = ExceptionUtils.convert(ex);
+                if (RECOVERY_PREDICATE.test(amqpException)) {
                   LOGGER.debug(
                       "Error during topology recovery, queueing recovery task for '{}', error is {}",
                       this.name(),
@@ -424,7 +428,10 @@ private void recoverAfterConnectionFailure(
                           () -> {
                             if (!this.recoveringConnection.get()) {
                               recoverAfterConnectionFailure(
-                                  recoveryConfiguration, name, ex, disconnectedHandlerReference);
+                                  recoveryConfiguration,
+                                  name,
+                                  amqpException,
+                                  disconnectedHandlerReference);
                             }
                           });
                 }
@@ -478,7 +485,7 @@ public <T> T maybeRetry(Supplier<T> task) {
                       }
                     },
                     connectionName))
-        .description("Trying to recover native connection for '%s'.", connectionName)
+        .description("Recovering native connection for '%s'.", connectionName)
         .delayPolicy(recoveryConfiguration.backOffDelayPolicy())
         .retry(RECOVERY_PREDICATE)
         .scheduler(this.scheduledExecutorService())
@@ -582,9 +589,9 @@ private void closeManagement() {
     this.management.close();
   }
 
-  private void releaseManagementResources() {
+  private void releaseManagementResources(AmqpException e) {
     if (this.management != null) {
-      this.management.releaseResources();
+      this.management.releaseResources(e);
     }
   }
 

src/main/java/com/rabbitmq/client/amqp/impl/AmqpManagement.java

@@ -202,7 +202,7 @@ public void close() {
     }
     if (this.closed.compareAndSet(false, true)) {
       this.state(CLOSED);
-      this.releaseResources();
+      this.releaseResources(null);
       if (this.receiver != null) {
         try {
           this.receiver.close();
@@ -361,12 +361,13 @@ private void failRequests(AmqpException exception) {
     }
   }
 
-  void releaseResources() {
+  void releaseResources(AmqpException e) {
     this.markUnavailable();
     if (this.receiveLoop != null) {
       this.receiveLoop.cancel(true);
       this.receiveLoop = null;
     }
+    this.failRequests(e);
   }
 
   QueueInfo declareQueue(String name, Map<String, Object> body) {
@@ -599,6 +600,7 @@ private OutstandingRequest(Duration timeout) {
 
     void block() {
       boolean completed;
+      long start = System.nanoTime();
       try {
         completed = this.latch.await(timeout.toMillis(), MILLISECONDS);
       } catch (InterruptedException e) {
@@ -609,7 +611,8 @@ void block() {
         throw this.exception.get();
       }
       if (!completed) {
-        throw new AmqpException("Could not get management response in %d ms", timeout.toMillis());
+        Duration duration = Duration.ofNanos(System.nanoTime() - start);
+        throw new AmqpException("Could not get management response in %d ms", duration.toMillis());
       }
     }
 

src/main/java/com/rabbitmq/client/amqp/impl/ConnectionUtils.java

@@ -115,7 +115,7 @@ static AmqpConnection.NativeConnectionWrapper enforceAffinity(
                 pickedConnection = connectionWrapper;
               } else {
                 LOGGER.debug("Affinity no longer valid, retrying.");
-                management.releaseResources();
+                management.releaseResources(null);
                 connectionWrapper.connection().close();
               }
             }
@@ -143,13 +143,13 @@ static AmqpConnection.NativeConnectionWrapper enforceAffinity(
               queueInfoRefreshed = true;
             }
           }
-          management.releaseResources();
+          management.releaseResources(null);
           connectionWrapper.connection().close();
         }
       }
       return pickedConnection;
     } catch (AmqpException.AmqpConnectionException e) {
-      management.releaseResources();
+      management.releaseResources(e);
       try {
         if (connectionWrapper != null) {
           connectionWrapper.connection().close();

src/main/java/com/rabbitmq/client/amqp/impl/ExceptionUtils.java

@@ -49,7 +49,9 @@ static <T> T wrapGet(Future<T> future) throws ClientException {
   }
 
   static AmqpException convert(Exception e) {
-    if (e instanceof ClientException) {
+    if (e instanceof AmqpException) {
+      return (AmqpException) e;
+    } else if (e instanceof ClientException) {
       return convert((ClientException) e);
     } else {
       return new AmqpException(e);