Add simple test with rolling cluster restart

Author: acogoluegnes

.github/workflows/test-pr.yml

@@ -41,6 +41,6 @@ jobs:
       - name: Start cluster
         run: ci/start-cluster.sh
       - name: Test against cluster
-        run: ./mvnw test -Dtest=ClusterTest -Drabbitmqctl.bin=DOCKER:rabbitmq0
+        run: ./mvnw test -Dtest="*ClusterTest" -Drabbitmqctl.bin=DOCKER:rabbitmq0
       - name: Stop cluster
         run: docker compose --file ci/cluster/docker-compose.yml down

.github/workflows/test-supported-java-versions.yml

@@ -49,6 +49,6 @@ jobs:
       - name: Start cluster
         run: ci/start-cluster.sh
       - name: Test against cluster
-        run: ./mvnw test -Dtest=ClusterTest -Drabbitmqctl.bin=DOCKER:rabbitmq0
+        run: ./mvnw test -Dtest="*ClusterTest" -Drabbitmqctl.bin=DOCKER:rabbitmq0
       - name: Stop cluster
         run: docker compose --file ci/cluster/docker-compose.yml down

.github/workflows/test.yml

@@ -47,7 +47,7 @@ jobs:
       - name: Start cluster
         run: ci/start-cluster.sh
       - name: Test against cluster
-        run: ./mvnw test -Dtest=ClusterTest -Drabbitmqctl.bin=DOCKER:rabbitmq0
+        run: ./mvnw test -Dtest="*ClusterTest" -Drabbitmqctl.bin=DOCKER:rabbitmq0
       - name: Stop cluster
         run: docker compose --file ci/cluster/docker-compose.yml down
       - name: Publish snapshot

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

@@ -44,12 +44,12 @@ private Cli() {}
       Map.of(
           "rabbit@node0", "rabbitmq0",
           "rabbit@node1", "rabbitmq1",
-          "rabbit@node2", "rabbitmq3");
+          "rabbit@node2", "rabbitmq2");
 
   public static String rabbitmqctlCommand() {
     String rabbitmqCtl = System.getProperty("rabbitmqctl.bin");
     if (rabbitmqCtl == null) {
-      throw new IllegalStateException("Please define the rabbitmqctl.bin system property");
+      rabbitmqCtl = "DOCKER:rabbitmq";
     }
     if (rabbitmqCtl.startsWith(DOCKER_PREFIX)) {
       String containerId = rabbitmqCtl.split(":")[1];
@@ -59,7 +59,7 @@ public static String rabbitmqctlCommand() {
     }
   }
 
-  static String rabbitmqQueuesCommand() {
+  private static String rabbitmqQueuesCommand() {
     String rabbitmqctl = rabbitmqctlCommand();
     int lastIndex = rabbitmqctl.lastIndexOf("rabbitmqctl");
     if (lastIndex == -1) {
@@ -68,6 +68,15 @@ static String rabbitmqQueuesCommand() {
     return rabbitmqctl.substring(0, lastIndex) + "rabbitmq-queues";
   }
 
+  private static String rabbitmqUpgradeCommand() {
+    String rabbitmqctl = rabbitmqctlCommand();
+    int lastIndex = rabbitmqctl.lastIndexOf("rabbitmqctl");
+    if (lastIndex == -1) {
+      throw new IllegalArgumentException("Not a valid rabbitqmctl command: " + rabbitmqctl);
+    }
+    return rabbitmqctl.substring(0, lastIndex) + "rabbitmq-upgrade";
+  }
+
   static String rabbitmqStreamsCommand() {
     String rabbitmqctl = rabbitmqctlCommand();
     int lastIndex = rabbitmqctl.lastIndexOf("rabbitmqctl");
@@ -89,6 +98,10 @@ static ProcessState rabbitmqStreams(String command) {
     return executeCommand(rabbitmqStreamsCommand() + " " + command);
   }
 
+  static ProcessState rabbitmqUpgrade(String command) {
+    return executeCommand(rabbitmqUpgradeCommand() + " " + command);
+  }
+
   static ProcessState rabbitmqctlIgnoreError(String command) {
     return executeCommand(rabbitmqctlCommand() + " " + command, true);
   }
@@ -236,9 +249,34 @@ static void pauseNode(String node) {
   }
 
   static void unpauseNode(String node) {
+    executeCommand("docker unpause " + nodeToDockerContainer(node));
+  }
+
+  static void restartNode(String node) {
+    String container = nodeToDockerContainer(node);
+    String dockerCommand = "docker exec " + container + " ";
+    String rabbitmqUpgradeCommand = dockerCommand + "rabbitmq-upgrade ";
+    executeCommand(rabbitmqUpgradeCommand + "await_online_quorum_plus_one -t 300");
+    executeCommand(rabbitmqUpgradeCommand + "drain");
+    executeCommand("docker stop " + container);
+    executeCommand("docker start " + container);
+    String otherContainer =
+        DOCKER_NODES_TO_CONTAINERS.values().stream()
+            .filter(c -> !c.endsWith(container))
+            .findAny()
+            .get();
+    executeCommand(
+        "docker exec "
+            + otherContainer
+            + " rabbitmqctl await_online_nodes "
+            + DOCKER_NODES_TO_CONTAINERS.size());
+    executeCommand(dockerCommand + "rabbitmqctl status");
+  }
+
+  private static String nodeToDockerContainer(String node) {
     String containerId = DOCKER_NODES_TO_CONTAINERS.get(node);
     Assert.notNull(containerId, "No container for node " + node);
-    executeCommand("docker unpause " + containerId);
+    return containerId;
   }
 
   static List<ConnectionInfo> listConnections() {

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

@@ -0,0 +1,83 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.client.amqp.impl;
+
+import static com.rabbitmq.client.amqp.impl.TestUtils.*;
+import static java.time.Duration.ofSeconds;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.rabbitmq.client.amqp.*;
+import java.util.List;
+import org.junit.jupiter.api.*;
+
+@DisabledIfNotCluster
+public class RecoveryClusterTest {
+
+  static final String[] URIS =
+      new String[] {"amqp://localhost:5672", "amqp://localhost:5673", "amqp://localhost:5674"};
+  static final BackOffDelayPolicy BACK_OFF_DELAY_POLICY = BackOffDelayPolicy.fixed(ofSeconds(1));
+  Environment environment;
+  Connection connection;
+  Management management;
+
+  @BeforeEach
+  void init() {
+    environment =
+        new AmqpEnvironmentBuilder().connectionSettings().uris(URIS).environmentBuilder().build();
+    this.connection =
+        environment
+            .connectionBuilder()
+            .recovery()
+            .backOffDelayPolicy(BACK_OFF_DELAY_POLICY)
+            .connectionBuilder()
+            .build();
+    this.management = connection.management();
+  }
+
+  @AfterEach
+  void tearDown() {
+    environment.close();
+  }
+
+  @Test
+  void clusterRestart(TestInfo info) {
+    String name = TestUtils.name(info);
+    try {
+      Management.QueueInfo queueInfo =
+          management.queue(name).type(Management.QueueType.QUORUM).declare();
+      List<String> replicas = queueInfo.replicas();
+      replicas.forEach(Cli::restartNode);
+
+      waitAtMost(
+          () -> {
+            try {
+              management.queueInfo(name);
+              return true;
+            } catch (Exception e) {
+              return false;
+            }
+          });
+      assertThat(management.queueInfo(name).replicas())
+          .hasSameSizeAs(replicas)
+          .containsExactlyInAnyOrderElementsOf(replicas);
+
+    } finally {
+      management.queueDeletion().delete(name);
+    }
+  }
+}