Add test for consumer with moving quorum queue

Author: acogoluegnes

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

@@ -21,9 +21,11 @@
 
 import com.rabbitmq.client.amqp.Management;
 import java.time.Duration;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 import org.assertj.core.api.AbstractObjectAssert;
 
 final class Assertions {
@@ -218,6 +220,7 @@ private ConnectionAssert(AmqpConnection connection) {
     }
 
     ConnectionAssert hasNodename(String nodename) {
+      Assert.notNull(nodename, "Expected nodename cannot be null");
       isNotNull();
       if (!actual.connectionNodename().equals(nodename)) {
         fail(
@@ -226,5 +229,19 @@ ConnectionAssert hasNodename(String nodename) {
       }
       return this;
     }
+
+    ConnectionAssert isOnFollower(Management.QueueInfo info) {
+      Assert.notNull(info, "Queue info cannot be null");
+      List<String> followers =
+          info.replicas().stream()
+              .filter(n -> !n.equals(info.leader()))
+              .collect(Collectors.toList());
+      if (!followers.contains(actual.connectionNodename())) {
+        fail(
+            "Connection is expected to be on follower node(s) '%s' but is on '%s'",
+            followers, actual.connectionNodename());
+      }
+      return this;
+    }
   }
 }

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

@@ -148,7 +148,64 @@ void publishToMovingQq() {
               })
           .build();
       assertThat(consumeSync).completes();
-      assertThat(messageIds).containsOnly(1L, 2L, 3L);
+      assertThat(messageIds).containsExactlyInAnyOrder(1L, 2L, 3L);
+    } finally {
+      management.queueDeletion().delete(q);
+    }
+  }
+
+  @Test
+  void consumeFromMovingQq() {
+    try {
+      management.queue(q).type(Management.QueueType.QUORUM).declare();
+
+      AmqpConnection consumeConnection = connection(b -> b.affinity().queue(q).operation(CONSUME));
+      assertThat(consumeConnection).isOnFollower(queueInfo());
+
+      Set<Long> messageIds = ConcurrentHashMap.newKeySet();
+      Sync consumeSync = sync();
+      consumeConnection
+          .consumerBuilder()
+          .queue(q)
+          .messageHandler(
+              (ctx, msg) -> {
+                messageIds.add(msg.messageIdAsLong());
+                consumeSync.down();
+                ctx.accept();
+              })
+          .build();
+
+      Publisher publisher = connection.publisherBuilder().queue(q).build();
+      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();
+
+      String follower = consumeConnection.connectionNodename();
+
+      deleteQqMember(follower);
+
+      publisher.publish(publisher.message().messageId(2L), ctx -> publishSync.down());
+      assertThat(publishSync).completes();
+      publishSync.reset();
+
+      assertThat(consumeSync).completes();
+      assertThat(messageIds).containsExactlyInAnyOrder(1L, 2L);
+      consumeSync.reset();
+
+      addQqMember(follower);
+
+      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();
     } finally {
       management.queueDeletion().delete(q);
     }
@@ -166,12 +223,16 @@ String deleteQqLeader() {
     Management.QueueInfo info = queueInfo();
     String initialLeader = info.leader();
     int initialReplicaCount = info.replicas().size();
-    Cli.deleteQuorumQueueMember(q, initialLeader);
+    deleteQqMember(initialLeader);
     TestUtils.waitAtMost(() -> !queueInfo().leader().equals(initialLeader));
     assertThat(queueInfo().replicas()).hasSize(initialReplicaCount - 1);
     return initialLeader;
   }
 
+  void deleteQqMember(String member) {
+    Cli.deleteQuorumQueueMember(q, member);
+  }
+
   void addQqMember(String newMember) {
     Management.QueueInfo info = queueInfo();
     int initialReplicaCount = info.replicas().size();