Recover also on resource invalid state exceptions

References #118

Author: acogoluegnes

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

@@ -47,8 +47,24 @@
 
 final class AmqpConnection extends ResourceBase implements Connection {
 
+  /** Connection-related issues */
+  private static final Predicate<Exception> CONNECTION_EXCEPTION_PREDICATE =
+      e -> e instanceof AmqpException.AmqpConnectionException;
+
+  /**
+   * Issues related to underlying resources.
+   *
+   * <p>E.g. the connection used for enforcing affinity gets closed, the management is marked as
+   * unavailable and throws an invalid state exception when it is called. The recovery process
+   * should restart.
+   */
+  private static final Predicate<Exception> RESOURCE_INVALID_STATE_PREDICATE =
+      e ->
+          e instanceof AmqpException.AmqpResourceInvalidStateException
+              && !(e instanceof AmqpException.AmqpResourceClosedException);
+
   private static final Predicate<Exception> RECOVERY_PREDICATE =
-      t -> t instanceof AmqpException.AmqpConnectionException;
+      CONNECTION_EXCEPTION_PREDICATE.or(RESOURCE_INVALID_STATE_PREDICATE);
 
   private static final AtomicLong ID_SEQUENCE = new AtomicLong(0);
 
@@ -399,7 +415,7 @@ private void recoverAfterConnectionFailure(
                     ex.getMessage());
                 if (RECOVERY_PREDICATE.test(ex)) {
                   LOGGER.debug(
-                      "Error during topology recoverable, queueing recovery task for '{}', error is {}",
+                      "Error during topology recovery, queueing recovery task for '{}', error is {}",
                       this.name(),
                       ex.getMessage());
                   this.environment

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

@@ -233,7 +233,7 @@ void init() {
           if (!this.initializing && this.state() != OPEN) {
             this.initializing = true;
             LOGGER.debug("Initializing management ({}).", this);
-            this.state(UNAVAILABLE);
+            this.markUnavailable();
             try {
               if (this.receiveLoop != null) {
                 this.receiveLoop.cancel(true);
@@ -313,7 +313,7 @@ private Runnable receiveTask() {
       } catch (ClientConnectionRemotelyClosedException | ClientLinkRemotelyClosedException e) {
         // receiver is closed
       } catch (ClientSessionRemotelyClosedException e) {
-        this.state(UNAVAILABLE);
+        this.markUnavailable();
         LOGGER.info("Management session closed in receive loop: {} ({})", e.getMessage(), this);
         AmqpException exception = ExceptionUtils.convert(e);
         this.failRequests(exception);

src/test/resources/logback-test.xml

@@ -18,7 +18,7 @@
     <logger name="com.rabbitmq.client.amqp.impl.AsyncRetry" level="warn" />
     <logger name="com.rabbitmq.client.amqp.impl.EntityRecovery" level="warn" />
     <logger name="com.rabbitmq.client.amqp.impl.EventLoop" level="warn" />
-    <logger name="com.rabbitmq.client.amqp.impl.AmqpManagement" level="warn" />
+    <logger name="com.rabbitmq.client.amqp.impl.AmqpManagement" level="info" />
     <logger name="com.rabbitmq.client.amqp.impl.AmqpPublisher" level="warn" />
     <logger name="com.rabbitmq.client.amqp.impl.AmqpConsumer" level="warn" />
     <logger name="com.rabbitmq.client.amqp.impl.RetryUtils" level="warn" />