Back to single thread for connection recovery

This does not solve the problem of recovery when there are too many
connections for the number of available processors. The sleep operations
between connection retry attempts block threads, so an approached based
on scheduling should be more appropriate.

Author: acogoluegnes

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

@@ -90,52 +90,27 @@ final class AmqpConnection extends ResourceBase implements Connection {
 
     if (recoveryConfiguration.activated()) {
       this.recoveryRequestQueue = new ArrayBlockingQueue<>(10);
-      Duration recoveryCheckPeriod = Duration.ofMillis(100L);
       this.recoveryTask =
-          this.scheduledExecutorService()
-              .scheduleAtFixedRate(
+          this.executorService()
+              .submit(
                   () -> {
-                    Runnable recoveryTask = this.recoveryRequestQueue.poll();
-                    if (recoveryTask != null) {
-                      this.executorService()
-                          .submit(
-                              () -> {
-                                try {
-                                  LOGGER.debug(
-                                      "Running recovery task for connection '{}'...", this.name());
-                                  recoveryTask.run();
-                                  LOGGER.debug(
-                                      "Recovery task for connection '{}' has run.", this.name());
-                                } catch (Exception e) {
-                                  LOGGER.warn(
-                                      "Error during connection recovery for '{}'", this.name(), e);
-                                }
-                              });
+                    LOGGER.debug("Starting recovery loop for '{}'", this.name());
+                    boolean keepGoing = true;
+                    while (keepGoing) {
+                      try {
+                        Runnable recoveryTask = this.recoveryRequestQueue.take();
+                        LOGGER.debug("Running recovery task for connection '{}'...", this.name());
+                        recoveryTask.run();
+                        LOGGER.debug("Recovery task for connection '{}' has run.", this.name());
+                      } catch (InterruptedException e) {
+                        LOGGER.info("Recovery loop for '{}' has been interrupted.", this.name());
+                        keepGoing = false;
+                      } catch (Exception e) {
+                        LOGGER.warn("Error during connection recovery for '{}'", this.name(), e);
+                      }
                     }
-                  },
-                  recoveryCheckPeriod.toMillis(),
-                  recoveryCheckPeriod.toMillis(),
-                  TimeUnit.MILLISECONDS);
-      //      this.recoveryLoop =
-      //          this.executorService()
-      //              .submit(
-      //                  () -> {
-      //                    while (!Thread.currentThread().isInterrupted()) {
-      //                      try {
-      //                        Runnable recoveryTask = this.recoveryRequestQueue.take();
-      //                        LOGGER.debug("Running recovery task for connection '{}'...",
-      // this.name());
-      //                        recoveryTask.run();
-      //                        LOGGER.debug("Recovery task for connection '{}' has run.",
-      // this.name());
-      //                      } catch (InterruptedException e) {
-      //                        return;
-      //                      } catch (Exception e) {
-      //                        LOGGER.warn("Error during connection recovery for '{}'",
-      // this.name(), e);
-      //                      }
-      //                    }
-      //                  });
+                    LOGGER.info("Closing recovery loop for '{}'.", this.name());
+                  });
       disconnectHandler = recoveryDisconnectHandler(recoveryConfiguration, builder.name());
     } else {
       disconnectHandler =
@@ -171,7 +146,8 @@ final class AmqpConnection extends ResourceBase implements Connection {
             this.affinity,
             this.environment.affinityCache(),
             this.affinityStrategy,
-            ConnectionUtils.NO_RETRY_STRATEGY);
+            ConnectionUtils.NO_RETRY_STRATEGY,
+            this.name());
     this.sync(ncw);
     LOGGER.debug("Opened connection '{}' on node '{}'.", this.name(), this.connectionNodename());
     this.state(OPEN);
@@ -259,17 +235,18 @@ private NativeConnectionWrapper connect(
     Address address = connectionSettings.selectAddress(addresses);
     StopWatch stopWatch = new StopWatch();
     try {
-      LOGGER.trace("Connecting...");
+      LOGGER.trace("Connecting '{}' to {}...", this.name(), address);
       org.apache.qpid.protonj2.client.Connection connection =
           this.environment.client().connect(address.host(), address.port(), connectionOptions);
+      LOGGER.debug("Created native connection instance for '{}'", this.name());
       ExceptionUtils.wrapGet(connection.openFuture());
-      LOGGER.trace("Connection attempt succeeded");
+      LOGGER.debug("Connection attempt '{}' succeeded", this.name());
       checkBrokerVersion(connection);
       return new NativeConnectionWrapper(connection, extractNode(connection), address);
     } catch (ClientException e) {
       throw ExceptionUtils.convert(e);
     } finally {
-      LOGGER.trace("Connection attempt took {}", stopWatch.stop());
+      LOGGER.debug("Connection attempt for '{}' took {}", this.name(), stopWatch.stop());
     }
   }
 
@@ -302,7 +279,9 @@ private static String extractNode(org.apache.qpid.protonj2.client.Connection con
   TopologyListener createTopologyListener(AmqpConnectionBuilder builder) {
     TopologyListener topologyListener;
     if (builder.recoveryConfiguration().topology()) {
-      RecordingTopologyListener rtl = new RecordingTopologyListener(this.executorService());
+      RecordingTopologyListener rtl =
+          new RecordingTopologyListener(
+              "topology-listener-connection-" + this.name(), this.executorService());
       this.entityRecovery = new EntityRecovery(this, rtl);
       topologyListener = rtl;
     } else {
@@ -346,9 +325,12 @@ TopologyListener createTopologyListener(AmqpConnectionBuilder builder) {
                 this.name(),
                 exception.getMessage());
             this.recoveryRequestQueue.add(
-                () ->
+                () -> {
+                  if (!this.recoveringConnection.get()) {
                     recoverAfterConnectionFailure(
-                        recoveryConfiguration, name, exception, resultReference));
+                        recoveryConfiguration, name, exception, resultReference);
+                  }
+                });
           } else {
             LOGGER.debug(
                 "Not recovering connection '{}' for error {}",
@@ -364,7 +346,7 @@ TopologyListener createTopologyListener(AmqpConnectionBuilder builder) {
   private void recoverAfterConnectionFailure(
       AmqpConnectionBuilder.AmqpRecoveryConfiguration recoveryConfiguration,
       String connectionName,
-      AmqpException failureCause,
+      Exception failureCause,
       AtomicReference<BiConsumer<org.apache.qpid.protonj2.client.Connection, DisconnectionEvent>>
           disconnectedHandlerReference) {
     LOGGER.info(
@@ -380,41 +362,59 @@ private void recoverAfterConnectionFailure(
     this.connectionAddress = null;
     LOGGER.debug("Releasing management resource of connection '{}'.", this.name());
     this.releaseManagementResources();
-    try {
-      this.recoveringConnection.set(true);
-      LOGGER.debug("Connection attempt for '{}'.", this.name());
-      NativeConnectionWrapper ncw =
-          recoverNativeConnection(
-              recoveryConfiguration, connectionName, disconnectedHandlerReference);
-      this.sync(ncw);
-      LOGGER.debug("Reconnected '{}' to {}", this.name(), this.currentConnectionLabel());
-    } catch (Exception ex) {
-      if (ex instanceof InterruptedException) {
-        Thread.currentThread().interrupt();
+    if (this.recoveringConnection.compareAndSet(false, true)) {
+      try {
+        this.recoveringConnection.set(true);
+        LOGGER.debug("Connection attempt for '{}'.", this.name());
+        NativeConnectionWrapper ncw =
+            recoverNativeConnection(
+                recoveryConfiguration, connectionName, disconnectedHandlerReference);
+        this.sync(ncw);
+        LOGGER.debug("Reconnected '{}' to {}", this.name(), this.currentConnectionLabel());
+      } catch (Exception ex) {
+        if (ex instanceof InterruptedException) {
+          Thread.currentThread().interrupt();
+        }
+        this.close(ex);
+        return;
+      } finally {
+        this.recoveringConnection.set(false);
       }
-      this.closed.set(true);
-      this.changeStateOfPublishers(CLOSED, ex);
-      this.changeStateOfConsumers(CLOSED, ex);
-      this.state(CLOSED, ex);
+    } else {
+      LOGGER.debug("Connection '{}' already recovering, returning.", this.name());
       return;
-    } finally {
-      this.recoveringConnection.set(false);
     }
 
     try {
       if (recoveryConfiguration.topology()) {
+        this.management.init();
         LOGGER.debug("Recovering topology of connection '{}'...", this.name());
         this.recoverTopology();
         this.recoverConsumers();
         this.recoverPublishers();
         LOGGER.debug("Recovered topology of connection '{}'.", this.name());
       }
-
-      LOGGER.info("Recovered connection to {}", this.currentConnectionLabel());
+      LOGGER.info("Recovered connection '{}' to {}", this.name(), this.currentConnectionLabel());
       this.state(OPEN);
     } catch (Exception ex) {
       // likely InterruptedException or IO exception
-      LOGGER.info("Error while trying to recover connection", ex);
+      LOGGER.info(
+          "Error while trying to recover topology for connection '{}': {}",
+          this.name(),
+          ex.getMessage());
+      if (RECOVERY_PREDICATE.test(ex)) {
+        LOGGER.debug(
+            "Error during topology recoverable, queueing recovery task for '{}', error is {}",
+            this.name(),
+            ex.getMessage());
+        this.recoveryRequestQueue.add(
+            () -> {
+              if (!this.recoveringConnection.get()) {
+                recoverAfterConnectionFailure(
+                    recoveryConfiguration, name, ex, disconnectedHandlerReference);
+              }
+            });
+      }
     }
   }
 
@@ -428,10 +428,14 @@ private NativeConnectionWrapper recoverNativeConnection(
     while (true) {
       Duration delay = recoveryConfiguration.backOffDelayPolicy().delay(attempt);
       if (BackOffDelayPolicy.TIMEOUT.equals(delay)) {
+        LOGGER.debug("Reached timeout to recover '{}'", this.name());
         throw new AmqpException("Recovery retry timed out");
       } else {
         try {
+          LOGGER.debug(
+              "Waiting for {} before trying to recover connection '{}'", delay, this.name());
           Thread.sleep(delay.toMillis());
+          LOGGER.debug("Done waiting for '{}' recovery.", this.name());
         } catch (InterruptedException ex) {
           Thread.currentThread().interrupt();
           LOGGER.info("Thread interrupted while waiting during connection recovery");
@@ -440,38 +444,40 @@ private NativeConnectionWrapper recoverNativeConnection(
       }
 
       try {
-        NativeConnectionWrapper result =
-            ConnectionUtils.enforceAffinity(
-                addrs -> {
-                  NativeConnectionWrapper wrapper =
-                      connect(
-                          this.connectionSettings,
-                          connectionName,
-                          disconnectedHandlerReference.get(),
-                          addrs);
-                  this.nativeConnection = wrapper.connection();
-                  return wrapper;
-                },
-                this.management,
-                this.affinity,
-                this.environment.affinityCache(),
-                this.affinityStrategy,
-                new ConnectionUtils.RetryStrategy() {
-                  @Override
-                  public <T> T maybeRetry(Supplier<T> task) {
-                    return RetryUtils.callAndMaybeRetry(
-                        task::get,
-                        e -> true,
-                        recoveryConfiguration.backOffDelayPolicy(),
-                        "Connection affinity operation");
-                  }
-                });
-        return result;
+        LOGGER.debug("Trying to create native connection for '{}'.", this.name());
+        return ConnectionUtils.enforceAffinity(
+            addrs -> {
+              NativeConnectionWrapper wrapper =
+                  connect(
+                      this.connectionSettings,
+                      connectionName,
+                      disconnectedHandlerReference.get(),
+                      addrs);
+              this.nativeConnection = wrapper.connection();
+              return wrapper;
+            },
+            this.management,
+            this.affinity,
+            this.environment.affinityCache(),
+            this.affinityStrategy,
+            new ConnectionUtils.RetryStrategy() {
+              @Override
+              public <T> T maybeRetry(Supplier<T> task) {
+                return RetryUtils.callAndMaybeRetry(
+                    task::get,
+                    e -> true,
+                    recoveryConfiguration.backOffDelayPolicy(),
+                    "Connection affinity operation");
+              }
+            },
+            this.name());
       } catch (Exception ex) {
         LOGGER.info("Error while trying to recover connection", ex);
         if (!RECOVERY_PREDICATE.test(ex)) {
           LOGGER.info(
-              "Stopping connection recovery, exception is not recoverable: {}", ex.getMessage());
+              "Stopping connection '{}' recovery, exception is not recoverable: {}",
+              this.name(),
+              ex.getMessage());
           throw new AmqpException("Could not recover connection after fatal exception", ex);
         }
       }
@@ -721,7 +727,10 @@ private void close(Throwable cause) {
         consumer.close();
       }
       try {
-        this.nativeConnection.close();
+        org.apache.qpid.protonj2.client.Connection nc = this.nativeConnection;
+        if (nc != null) {
+          nc.close();
+        }
       } catch (Exception e) {
         LOGGER.warn("Error while closing native connection", e);
       }

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

@@ -26,7 +26,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.qpid.protonj2.client.*;
-import org.apache.qpid.protonj2.client.futures.ClientFutureFactory;
 
 class AmqpEnvironment implements Environment {
 
@@ -56,7 +55,7 @@ class AmqpEnvironment implements Environment {
     this.id = ID_SEQUENCE.getAndIncrement();
     connectionSettings.copyTo(this.connectionSettings);
     this.connectionSettings.consolidate();
-    ClientOptions clientOptions = new ClientOptions().futureType(ClientFutureFactory.PROGRESSIVE);
+    ClientOptions clientOptions = new ClientOptions();
     this.client = Client.create(clientOptions);
 
     if (executorService == null) {