Make sure stream has replicas in test

Author: acogoluegnes

src/test/java/com/rabbitmq/stream/Cli.java

@@ -26,13 +26,9 @@
 import java.util.*;
 import java.util.concurrent.Callable;
 import java.util.stream.Collectors;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public class Cli {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(Cli.class);
-
   private static final String DOCKER_PREFIX = "DOCKER:";
 
   private static final Gson GSON = new Gson();
@@ -252,15 +248,6 @@ public static String rabbitmqctlCommand() {
     }
   }
 
-  private static String dockerContainer() {
-    if (rabbitmqctlCommand().startsWith("docker")) {
-      String rabbitmqCtl = System.getProperty("rabbitmqctl.bin");
-      return rabbitmqCtl.split(":")[1];
-    } else {
-      throw new IllegalStateException("Broker does not run on broker");
-    }
-  }
-
   private static String rabbitmqStreamsCommand() {
     String rabbitmqctl = rabbitmqctlCommand();
     int lastIndex = rabbitmqctl.lastIndexOf("rabbitmqctl");
@@ -314,7 +301,7 @@ private static void setResourceAlarm(String source) throws IOException {
     rabbitmqctl("eval 'rabbit_alarm:set_alarm({{resource_limit, " + source + ", node()}, []}).'");
   }
 
-  private static void clearResourceAlarm(String source) throws IOException {
+  private static void clearResourceAlarm(String source) {
     rabbitmqctl("eval 'rabbit_alarm:clear_alarm({resource_limit, " + source + ", node()}).'");
   }
 
@@ -363,12 +350,6 @@ public static void restartNode(String node) {
     executeCommand(dockerCommand + "rabbitmqctl status");
   }
 
-  public static void restartBrokerContainer() {
-    String container = dockerContainer();
-    executeCommand("docker stop " + container);
-    executeCommand("docker start " + container);
-  }
-
   public static void rebalance() {
     rabbitmqQueues("rebalance all");
   }

src/test/java/com/rabbitmq/stream/impl/LoadBalancerClusterTest.java

@@ -15,6 +15,7 @@
 package com.rabbitmq.stream.impl;
 
 import static com.rabbitmq.stream.impl.Assertions.assertThat;
+import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
 import static java.lang.Integer.parseInt;
 import static java.util.stream.Collectors.toSet;
 import static java.util.stream.IntStream.range;
@@ -79,7 +80,7 @@ void tearDown() throws Exception {
 
   @ParameterizedTest
   @ValueSource(booleans = {false, true})
-  void pickConsumersAmongCandidates(boolean forceReplica) {
+  void pickConsumersAmongCandidates(boolean forceReplica) throws Exception {
     int maxSubscriptionsPerClient = 2;
     int subscriptionCount = maxSubscriptionsPerClient * 10;
     try (ConsumersCoordinator c =
@@ -91,6 +92,10 @@ void pickConsumersAmongCandidates(boolean forceReplica) {
             forceReplica,
             Utils.brokerPicker())) {
 
+      waitAtMost(
+          () -> locator.metadata(stream).get(stream).hasReplicas(),
+          "Stream '%s' should have replicas");
+
       range(0, subscriptionCount)
           .forEach(
               ignored -> {

src/test/java/com/rabbitmq/stream/impl/TestUtils.java

@@ -44,10 +44,7 @@
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.vavr.Tuple2;
-import java.io.BufferedReader;
 import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
 import java.lang.annotation.Documented;
 import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;
@@ -97,6 +94,11 @@ public static Duration waitAtMost(CallableBooleanSupplier condition) throws Exce
     return waitAtMost(DEFAULT_CONDITION_TIMEOUT, condition, null);
   }
 
+  public static Duration waitAtMost(
+      CallableBooleanSupplier condition, String format, Object... args) throws Exception {
+    return waitAtMost(DEFAULT_CONDITION_TIMEOUT, condition, () -> String.format(format, args));
+  }
+
   public static Duration waitAtMost(CallableBooleanSupplier condition, Supplier<String> message)
       throws Exception {
     return waitAtMost(DEFAULT_CONDITION_TIMEOUT, condition, message);
@@ -384,16 +386,6 @@ static boolean isCluster() {
     return !content.replace("[", "").replace("]", "").trim().isEmpty();
   }
 
-  private static String capture(InputStream is) throws IOException {
-    BufferedReader br = new BufferedReader(new InputStreamReader(is));
-    String line;
-    StringBuilder buff = new StringBuilder();
-    while ((line = br.readLine()) != null) {
-      buff.append(line).append("\n");
-    }
-    return buff.toString();
-  }
-
   static <T> void forEach(Collection<T> in, CallableIndexConsumer<T> consumer) throws Exception {
     int count = 0;
     for (T t : in) {