Add some logging for recovery

Author: acogoluegnes

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

@@ -70,10 +70,12 @@ final class AmqpConnection extends ResourceBase implements Connection {
   private final Supplier<SessionHandler> sessionHandlerSupplier;
   private final ConnectionUtils.AffinityContext affinity;
   private final ConnectionSettings.AffinityStrategy affinityStrategy;
+  private final String name;
 
   AmqpConnection(AmqpConnectionBuilder builder) {
     super(builder.listeners());
     this.id = ID_SEQUENCE.getAndIncrement();
+    this.name = builder.name();
     this.environment = builder.environment();
     this.connectionSettings = builder.connectionSettings().consolidate();
     this.sessionHandlerSupplier =
@@ -95,11 +97,13 @@ final class AmqpConnection extends ResourceBase implements Connection {
                     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", e);
+                        LOGGER.warn("Error during connection recovery for '{}'", this.name(), e);
                       }
                     }
                   });
@@ -125,6 +129,7 @@ final class AmqpConnection extends ResourceBase implements Connection {
       this.affinityStrategy = null;
     }
     this.management = createManagement();
+    LOGGER.debug("Opening native connection for connection '{}'...", this.name());
     NativeConnectionWrapper ncw =
         ConnectionUtils.enforceAffinity(
             addrs -> {
@@ -139,6 +144,7 @@ final class AmqpConnection extends ResourceBase implements Connection {
             this.affinityStrategy,
             ConnectionUtils.NO_RETRY_STRATEGY);
     this.sync(ncw);
+    LOGGER.debug("Opened connection '{}' on node '{}'.", this.name(), this.connectionNodename());
     this.state(OPEN);
     this.environment.metricsCollector().openConnection();
   }
@@ -286,29 +292,36 @@ TopologyListener createTopologyListener(AmqpConnectionBuilder builder) {
     BiConsumer<org.apache.qpid.protonj2.client.Connection, DisconnectionEvent> result =
         (conn, event) -> {
           ClientIOException ioex = event.failureCause();
-          LOGGER.debug("Disconnect handler, error is the following:", ioex);
+          LOGGER.debug("Disconnect handler, error is the following: {}", ioex.getMessage());
           if (this.state() == OPENING) {
             LOGGER.debug("Connection is still opening, disconnect handler skipped");
             // the broker is not available when opening the connection
             // nothing to do in this listener
             return;
           }
           if (this.recoveringConnection.get()) {
-            LOGGER.debug("Filtering recovery task enqueueing, connection recovery in progress");
+            LOGGER.debug(
+                "Filtering recovery task enqueueing, connection recovery of '{}' in progress",
+                this.name());
             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("Queueing recovery task, error is {}", exception.getMessage());
+            LOGGER.debug(
+                "Queueing recovery task for '{}', error is {}",
+                this.name(),
+                exception.getMessage());
             this.recoveryRequestQueue.add(
                 () ->
                     recoverAfterConnectionFailure(
                         recoveryConfiguration, name, exception, resultReference));
           } else {
             LOGGER.debug(
-                "Not recovering connection for error {}", event.failureCause().getMessage());
+                "Not recovering connection '{}' for error {}",
+                this.name(),
+                event.failureCause().getMessage());
           }
         };
 
@@ -323,7 +336,9 @@ private void recoverAfterConnectionFailure(
       AtomicReference<BiConsumer<org.apache.qpid.protonj2.client.Connection, DisconnectionEvent>>
           disconnectedHandlerReference) {
     LOGGER.info(
-        "Connection to {} has been disconnected, trying to recover", this.currentConnectionLabel());
+        "Connection '{}' to '{}' has been disconnected, trying to recover",
+        this.name(),
+        this.currentConnectionLabel());
     this.state(RECOVERING, failureCause);
     this.changeStateOfPublishers(RECOVERING, failureCause);
     this.changeStateOfConsumers(RECOVERING, failureCause);
@@ -337,7 +352,7 @@ private void recoverAfterConnectionFailure(
           recoverNativeConnection(
               recoveryConfiguration, connectionName, disconnectedHandlerReference);
       this.sync(ncw);
-      LOGGER.debug("Reconnected to {}", this.currentConnectionLabel());
+      LOGGER.debug("Reconnected '{}' to {}", this.name(), this.currentConnectionLabel());
     } catch (Exception ex) {
       if (ex instanceof InterruptedException) {
         Thread.currentThread().interrupt();
@@ -353,11 +368,11 @@ private void recoverAfterConnectionFailure(
 
     try {
       if (recoveryConfiguration.topology()) {
-        LOGGER.debug("Recovering topology");
+        LOGGER.debug("Recovering topology of connection '{}'...", this.name());
         this.recoverTopology();
         this.recoverConsumers();
         this.recoverPublishers();
-        LOGGER.debug("Recovered topology");
+        LOGGER.debug("Recovered topology of connection '{}'.", this.name());
       }
 
       LOGGER.info("Recovered connection to {}", this.currentConnectionLabel());
@@ -631,6 +646,10 @@ String connectionNodename() {
     return this.connectionNodename;
   }
 
+  String name() {
+    return this.name == null ? "<no-name>" : this.name;
+  }
+
   ConnectionUtils.AffinityContext affinity() {
     return this.affinity;
   }

src/test/java/com/rabbitmq/client/amqp/impl/RecoveryClusterTest.java

@@ -36,6 +36,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
 import java.util.stream.IntStream;
 import org.junit.jupiter.api.*;
 import org.slf4j.Logger;
@@ -187,6 +188,13 @@ void clusterRestart() {
       System.out.println("Consumers:");
       consumerStates.forEach(
           p -> System.out.printf("  queue %s, is on member? %s%n", p.queue, p.isOnMember()));
+    } catch (Throwable e) {
+      LOGGER.info("Test failed with {}", e.getMessage());
+      Consumer<AmqpConnection> log = c -> LOGGER.info("Connection {}: {}", c.name(), c.state());
+      log.accept(this.connection);
+      publisherStates.forEach(s -> log.accept(s.connection));
+      consumerStates.forEach(s -> log.accept(s.connection));
+      throw e;
     } finally {
       publisherStates.forEach(PublisherState::close);
       consumerStates.forEach(ConsumerState::close);

src/test/resources/logback-test.xml

@@ -1,7 +1,7 @@
 <configuration>
     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
         <encoder>
-            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{0} - %msg%n</pattern>
         </encoder>
     </appender>
 
@@ -11,10 +11,10 @@
     <logger name="com.rabbitmq.client.amqp.impl.AmqpConnection" level="warn" />
     <logger name="com.rabbitmq.client.amqp.impl.AmqpConsumer" level="warn" />
     <logger name="com.rabbitmq.client.amqp.impl.AmqpManagement" level="warn" />
-    <logger name="org.apache.qpid" level="warn" />
+    <logger name="org.apache.qpid" level="error" />
     <logger name="com.rabbitmq.client.amqp.perf" level="info" />
-    <logger name="com.rabbitmq.client.amqp.impl.ConnectionUtils" level="debug" />
-    <logger name="com.rabbitmq.client.amqp.impl.AmqpConnection" level="warn" />
+    <logger name="com.rabbitmq.client.amqp.impl.ConnectionUtils" level="warn" />
+    <logger name="com.rabbitmq.client.amqp.impl.AmqpConnection" level="debug" />
     <logger name="com.rabbitmq.client.amqp.impl.RecoveryClusterTest" level="info" />
 
     <root level="warn">