Merge pull request #83 from rabbitmq/fix-stream-consumer-recovery

Recover connection if consumer recovery closes it

Author: acogoluegnes

.github/workflows/test-pr.yml

@@ -24,8 +24,6 @@ jobs:
           cache: 'maven'
       - name: Start broker
         run: ci/start-broker.sh
-        env:
-          RABBITMQ_IMAGE: 'pivotalrabbitmq/rabbitmq:main'
       - name: Start toxiproxy
         run: ci/start-toxiproxy.sh
       - name: Display Java version

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

@@ -388,7 +388,7 @@ private void recoverAfterConnectionFailure(
                 this.state(OPEN);
               } catch (Exception ex) {
                 // likely InterruptedException or IO exception
-                LOGGER.info(
+                LOGGER.warn(
                     "Error while trying to recover topology for connection '{}': {}",
                     this.name(),
                     ex.getMessage());
@@ -481,6 +481,13 @@ private void recoverConsumers() throws InterruptedException {
           consumer.recoverAfterConnectionFailure();
           consumer.state(OPEN);
           LOGGER.debug("Recovered consumer {} (queue '{}')", consumer.id(), consumer.queue());
+        } catch (AmqpException.AmqpConnectionException ex) {
+          LOGGER.warn(
+              "Connection error while trying to recover consumer {} (queue '{}'), restarting recovery",
+              consumer.id(),
+              consumer.queue(),
+              ex);
+          throw ex;
         } catch (Exception ex) {
           LOGGER.warn(
               "Error while trying to recover consumer {} (queue '{}')",