Add test to consume from quorum queue while leader is paused

Author: acogoluegnes

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

@@ -27,6 +27,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.Callable;
 import java.util.function.Predicate;
 import java.util.regex.Matcher;
@@ -39,6 +40,11 @@ private Cli() {}
   private static final Pattern CONNECTION_NAME_PATTERN =
       Pattern.compile("\"connection_name\",\"(?<name>[a-zA-Z0-9_\\-]+)?\"");
   private static final String DOCKER_PREFIX = "DOCKER:";
+  private static final Map<String, String> DOCKER_NODES_TO_CONTAINERS =
+      Map.of(
+          "rabbit@node0", "rabbitmq0",
+          "rabbit@node1", "rabbitmq1",
+          "rabbit@node2", "rabbitmq3");
 
   public static String rabbitmqctlCommand() {
     String rabbitmqCtl = System.getProperty("rabbitmqctl.bin");
@@ -223,6 +229,18 @@ static void restartStream(String stream) {
     rabbitmqStreams(" restart_stream " + stream);
   }
 
+  static void pauseNode(String node) {
+    String containerId = DOCKER_NODES_TO_CONTAINERS.get(node);
+    Assert.notNull(containerId, "No container for node " + node);
+    executeCommand("docker pause " + containerId);
+  }
+
+  static void unpauseNode(String node) {
+    String containerId = DOCKER_NODES_TO_CONTAINERS.get(node);
+    Assert.notNull(containerId, "No container for node " + node);
+    executeCommand("docker unpause " + containerId);
+  }
+
   static List<ConnectionInfo> listConnections() {
     String output = rabbitmqctl("list_connections -q pid peer_port client_properties").output();
     // output (header line presence depends on broker version):

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

@@ -32,6 +32,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
 import org.junit.jupiter.api.*;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
@@ -213,7 +214,7 @@ void consumeFromMovingQq() {
   }
 
   @Test
-  void publishConsumeQqWhenLeaderChanges() {
+  void publishConsumeQuorumQueueWhenLeaderChanges() {
     try {
       management.queue(q).type(Management.QueueType.QUORUM).declare();
 
@@ -271,6 +272,80 @@ void publishConsumeQqWhenLeaderChanges() {
     }
   }
 
+  @Test
+  void consumeFromQuorumQueueWhenLeaderIsPaused() {
+    management.queue(q).type(Management.QueueType.QUORUM).declare();
+    Management.QueueInfo queueInfo = queueInfo();
+    String initialLeader = queueInfo.leader();
+    boolean nodePaused = false;
+    try {
+      AmqpConnection consumeConnection = connection(b -> b.affinity().queue(q).operation(CONSUME));
+      assertThat(consumeConnection).isOnFollower(queueInfo());
+      Management mgmt = consumeConnection.management();
+
+      Set<Long> messageIds = ConcurrentHashMap.newKeySet();
+      Sync consumeSync = sync();
+      consumeConnection
+          .consumerBuilder()
+          .queue(q)
+          .messageHandler(
+              (ctx, msg) -> {
+                messageIds.add(msg.messageIdAsLong());
+                consumeSync.down();
+                ctx.accept();
+              })
+          .build();
+
+      AmqpConnection publishConnection = connection(b -> b.affinity().queue(q).operation(CONSUME));
+      Publisher publisher = publishConnection.publisherBuilder().queue(q).build();
+      assertThat(publishConnection).isOnFollower(queueInfo());
+
+      Sync publishSync = sync();
+      publisher.publish(publisher.message().messageId(1L), ctx -> publishSync.down());
+      assertThat(publishSync).completes();
+      publishSync.reset();
+
+      assertThat(consumeSync).completes();
+      assertThat(messageIds).containsExactlyInAnyOrder(1L);
+      consumeSync.reset();
+
+      List<String> initialFollowers =
+          queueInfo.replicas().stream()
+              .filter(n -> !n.equals(initialLeader))
+              .collect(Collectors.toList());
+      assertThat(initialFollowers).isNotEmpty();
+
+      Cli.pauseNode(initialLeader);
+      nodePaused = true;
+
+      publisher.publish(publisher.message().messageId(2L), ctx -> publishSync.down());
+      assertThat(publishSync).completes();
+      publishSync.reset();
+
+      assertThat(consumeSync).completes();
+      assertThat(messageIds).containsExactlyInAnyOrder(1L, 2L);
+      consumeSync.reset();
+
+      Cli.unpauseNode(initialLeader);
+      nodePaused = false;
+
+      publisher.publish(publisher.message().messageId(3L), ctx -> publishSync.down());
+      assertThat(publishSync).completes();
+      publishSync.reset();
+
+      assertThat(consumeSync).completes();
+      assertThat(messageIds).containsExactlyInAnyOrder(1L, 2L, 3L);
+      consumeSync.reset();
+
+      waitAtMost(() -> initialFollowers.contains(mgmt.queueInfo(q).leader()));
+    } finally {
+      if (nodePaused) {
+        Cli.unpauseNode(initialLeader);
+      }
+      management.queueDeletion().delete(q);
+    }
+  }
+
   @Test
   void publishToRestartedStream() {
     try {