Add test for blocked connection

And make sure channel shutdown listener is called.

Author: acogoluegnes

src/test/java/com/rabbitmq/client/test/BlockedConnectionTest.java

@@ -1,4 +1,5 @@
-// Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+// Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom
+// Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -18,11 +19,27 @@
 import static com.rabbitmq.client.test.TestUtils.LatchConditions.completed;
 import static com.rabbitmq.client.test.TestUtils.waitAtMost;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.fail;
 
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
+
+import java.io.IOException;
+import java.time.Duration;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import com.rabbitmq.client.ConsumerShutdownSignalCallback;
+import com.rabbitmq.client.DeliverCallback;
+import com.rabbitmq.client.Delivery;
+import com.rabbitmq.client.MessageProperties;
+import com.rabbitmq.client.ShutdownListener;
+import com.rabbitmq.client.ShutdownSignalException;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
@@ -59,4 +76,34 @@ void errorInBlockListenerShouldCloseConnection(boolean nio) throws Exception {
     waitAtMost(() -> !c.isOpen());
   }
 
+  @Test
+  void shutdownListenerShouldBeCalledWhenChannelDies() throws Exception {
+    long confirmTimeout = Duration.ofSeconds(2).toMillis();
+    ConnectionFactory cf = TestUtils.connectionFactory();
+    Connection c = cf.newConnection();
+    boolean blocked = false;
+    try {
+      CountDownLatch blockedLatch = new CountDownLatch(1);
+      c.addBlockedListener(reason -> blockedLatch.countDown(), () -> {});
+      Channel ch = c.createChannel();
+      CountDownLatch chShutdownLatch = new CountDownLatch(1);
+      ch.addShutdownListener(cause -> chShutdownLatch.countDown());
+      ch.confirmSelect();
+      ch.basicPublish("", "", MessageProperties.BASIC, "".getBytes());
+      ch.waitForConfirmsOrDie(confirmTimeout);
+      block();
+      blocked = true;
+      ch.basicPublish("", "", MessageProperties.BASIC, "".getBytes());
+      assertThat(blockedLatch).is(completed());
+      ch.basicPublish("", "", MessageProperties.BASIC, "".getBytes());
+      assertThatThrownBy(() -> ch.waitForConfirmsOrDie(confirmTimeout))
+          .isInstanceOf(TimeoutException.class);
+      assertThat(chShutdownLatch).is(completed());
+    } finally {
+      if (blocked) {
+        unblock();
+      }
+      c.close();
+    }
+  }
 }

src/test/java/com/rabbitmq/client/test/BrokerTestCase.java

@@ -315,12 +315,12 @@ protected void clearResourceAlarm(String source) throws IOException {
         Host.clearResourceAlarm(source);
     }
 
-    protected void block() throws IOException, InterruptedException {
+    protected void block() throws IOException {
         Host.rabbitmqctl("set_vm_memory_high_watermark 0.000000001");
         setResourceAlarm("disk");
     }
 
-    protected void unblock() throws IOException, InterruptedException {
+    protected void unblock() throws IOException {
         Host.rabbitmqctl("set_vm_memory_high_watermark 0.4");
         clearResourceAlarm("disk");
     }

src/test/java/com/rabbitmq/client/test/LambdaCallbackTest.java

@@ -39,11 +39,7 @@ protected void createResources() throws IOException, TimeoutException {
     @Override
     protected void releaseResources() throws IOException {
         channel.queueDelete(queue);
-        try {
-            unblock();
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
+        unblock();
     }
 
     @Test public void shutdownListener() throws Exception {
@@ -78,13 +74,7 @@ protected void releaseResources() throws IOException {
         final CountDownLatch latch = new CountDownLatch(1);
         try(Connection connection = TestUtils.connectionFactory().newConnection()) {
             connection.addBlockedListener(
-                reason -> {
-                    try {
-                        unblock();
-                    } catch (InterruptedException e) {
-                        e.printStackTrace();
-                    }
-            },
+                reason -> unblock(),
                 () -> latch.countDown()
             );
             block();

src/test/java/com/rabbitmq/client/test/server/BlockedConnection.java

@@ -35,11 +35,7 @@
 
 public class BlockedConnection extends BrokerTestCase {
     protected void releaseResources() throws IOException {
-        try {
-            unblock();
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
+        unblock();
     }
     // this test first opens a connection, then triggers
     // and alarm and blocks
@@ -79,14 +75,10 @@ private Connection connection(final CountDownLatch latch) throws IOException, Ti
         Connection connection = factory.newConnection();
         connection.addBlockedListener(new BlockedListener() {
             public void handleBlocked(String reason) throws IOException {
-                try {
-                    unblock();
-                } catch (InterruptedException e) {
-                    e.printStackTrace();
-                }
+                unblock();
             }
 
-            public void handleUnblocked() throws IOException {
+            public void handleUnblocked() {
                 latch.countDown();
             }
         });

src/test/java/com/rabbitmq/tools/Host.java

@@ -254,10 +254,7 @@ public static String node_portB()
     }
 
     public static String rabbitmqctlCommand() {
-        String rabbitmqCtl = System.getProperty("rabbitmqctl.bin");
-        if (rabbitmqCtl == null) {
-            throw new IllegalStateException("Please define the rabbitmqctl.bin system property");
-        }
+        String rabbitmqCtl = rabbitmqctl();
         if (rabbitmqCtl.startsWith(DOCKER_PREFIX)) {
             String containerId = rabbitmqCtl.split(":")[1];
             return "docker exec " + containerId + " rabbitmqctl";
@@ -266,11 +263,12 @@ public static String rabbitmqctlCommand() {
         }
     }
 
+    private static String rabbitmqctl() {
+        return System.getProperty("rabbitmqctl.bin", "DOCKER:rabbitmq");
+    }
+
     public static boolean isOnDocker() {
-        String rabbitmqCtl = System.getProperty("rabbitmqctl.bin");
-        if (rabbitmqCtl == null) {
-            throw new IllegalStateException("Please define the rabbitmqctl.bin system property");
-        }
+        String rabbitmqCtl = rabbitmqctl();
         return rabbitmqCtl.startsWith(DOCKER_PREFIX);
     }