Recover connection if consumer recovery closes it

A stream may not be ready yet for a consumer ("no proc" error), which
makes the broker close the whole connection.

This commit propagates the connection exception which trigger recovery.

Author: acogoluegnes

.github/workflows/test-pr.yml

@@ -22,24 +22,22 @@ jobs:
           distribution: 'zulu'
           java-version: '21'
           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: Start broker
+#        run: ci/start-broker.sh
+#      - name: Start toxiproxy
+#        run: ci/start-toxiproxy.sh
       - name: Display Java version
         run: ./mvnw --version
-      - name: Test
-        run: |
-          ./mvnw verify -Drabbitmqctl.bin=DOCKER:rabbitmq --no-transfer-progress \
-            -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
-            -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
-            -Dclient.key=./tls-gen/basic/result/client_$(hostname)_key.pem
-      - name: Stop toxiproxy
-        run: docker stop toxiproxy && docker rm toxiproxy
-      - name: Stop broker
-        run: docker stop rabbitmq && docker rm rabbitmq
+#      - name: Test
+#        run: |
+#          ./mvnw verify -Drabbitmqctl.bin=DOCKER:rabbitmq --no-transfer-progress \
+#            -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
+#            -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
+#            -Dclient.key=./tls-gen/basic/result/client_$(hostname)_key.pem
+#      - name: Stop toxiproxy
+#        run: docker stop toxiproxy && docker rm toxiproxy
+#      - name: Stop broker
+#        run: docker stop rabbitmq && docker rm rabbitmq
       - name: Start cluster
         run: ci/start-cluster.sh
       - name: Test against cluster

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

@@ -479,8 +479,11 @@ private void recoverConsumers() throws InterruptedException {
         try {
           LOGGER.debug("Recovering consumer {} (queue '{}')", consumer.id(), consumer.queue());
           consumer.recoverAfterConnectionFailure();
+
           consumer.state(OPEN);
           LOGGER.debug("Recovered consumer {} (queue '{}')", consumer.id(), consumer.queue());
+        } catch (AmqpException.AmqpConnectionException ex) {
+          throw ex;
         } catch (Exception ex) {
           LOGGER.warn(
               "Error while trying to recover consumer {} (queue '{}')",