Activate Khepri in CI broker

Author: acogoluegnes

ci/start-broker.sh

@@ -45,5 +45,6 @@ docker run -d --name rabbitmq \
 
 wait_for_message rabbitmq "completed with"
 
+docker exec rabbitmq rabbitmqctl enable_feature_flag khepri_db
 docker exec rabbitmq rabbitmq-diagnostics erlang_version
 docker exec rabbitmq rabbitmqctl version

ci/start-cluster.sh

@@ -17,4 +17,8 @@ wait_for_message rabbitmq0 "completed with"
 
 docker exec rabbitmq0 rabbitmqctl await_online_nodes 3
 
+docker exec rabbitmq0 rabbitmqctl enable_feature_flag khepri_db
+docker exec rabbitmq1 rabbitmqctl enable_feature_flag khepri_db
+docker exec rabbitmq2 rabbitmqctl enable_feature_flag khepri_db
+
 docker exec rabbitmq0 rabbitmqctl cluster_status

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

@@ -23,6 +23,7 @@
 import static com.rabbitmq.client.amqp.impl.TestUtils.sync;
 import static com.rabbitmq.client.amqp.impl.TestUtils.waitAtMost;
 import static java.time.Duration.ofMillis;
+import static java.time.Duration.ofSeconds;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.rabbitmq.client.amqp.*;
@@ -37,7 +38,7 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
-@TestUtils.DisabledIfNotCluster
+// @TestUtils.DisabledIfNotCluster
 public class ClusterTest {
 
   static final BackOffDelayPolicy BACK_OFF_DELAY_POLICY = BackOffDelayPolicy.fixed(ofMillis(100));
@@ -319,13 +320,16 @@ void consumeFromQuorumQueueWhenLeaderIsPaused() {
       nodePaused = true;
 
       publisher.publish(publisher.message().messageId(2L), ctx -> publishSync.down());
-      assertThat(publishSync).completes();
+
+      assertThat(publishSync).completes(ofSeconds(20));
       publishSync.reset();
 
       assertThat(consumeSync).completes();
       assertThat(messageIds).containsExactlyInAnyOrder(1L, 2L);
       consumeSync.reset();
 
+      assertThat(initialFollowers).contains(mgmt.queueInfo(q).leader());
+
       Cli.unpauseNode(initialLeader);
       nodePaused = false;
 
@@ -335,7 +339,6 @@ void consumeFromQuorumQueueWhenLeaderIsPaused() {
 
       assertThat(consumeSync).completes();
       assertThat(messageIds).containsExactlyInAnyOrder(1L, 2L, 3L);
-      consumeSync.reset();
 
       waitAtMost(() -> initialFollowers.contains(mgmt.queueInfo(q).leader()));
     } finally {
@@ -451,7 +454,7 @@ String deleteLeader(Consumer<String> deleteMemberOperation) {
     String initialLeader = info.leader();
     int initialReplicaCount = info.replicas().size();
     deleteMemberOperation.accept(initialLeader);
-    TestUtils.waitAtMost(() -> !queueInfo().leader().equals(initialLeader));
+    TestUtils.waitAtMost(() -> !initialLeader.equals(queueInfo().leader()));
     assertThat(queueInfo().replicas()).hasSize(initialReplicaCount - 1);
     return initialLeader;
   }