Check all connections are there in tests

Before trying to kill some of them. Test fails on CI sometimes.

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/StreamConsumer.java

@@ -28,17 +28,17 @@
 import com.rabbitmq.stream.impl.Client.QueryOffsetResponse;
 import com.rabbitmq.stream.impl.StreamConsumerBuilder.TrackingConfiguration;
 import com.rabbitmq.stream.impl.StreamEnvironment.TrackingConsumerRegistration;
-import java.util.Objects;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
 import com.rabbitmq.stream.impl.Utils.CompositeConsumerUpdateListener;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.LongConsumer;
 import java.util.function.LongSupplier;

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

@@ -112,7 +112,7 @@ public static Process killConnection(String connectionName) throws IOException {
     return rabbitmqctl("eval 'rabbit_stream:kill_connection(\"" + connectionName + "\").'");
   }
 
-  private static List<ConnectionInfo> listConnections() throws IOException {
+  public static List<ConnectionInfo> listConnections() throws IOException {
     Process process =
         rabbitmqctl("list_stream_connections --formatter json conn_name,client_properties");
     return toConnectionInfoList(capture(process.getInputStream()));

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

@@ -408,6 +408,9 @@ void singleActiveConsumersShouldSpreadOnSuperStreamPartitions(TestInfo info) thr
   void killingConnectionsShouldTriggerConsumerUpdateNotification() throws Exception {
     Map<String, Boolean> consumerStates = new ConcurrentHashMap<>();
     List<String> consumerNames = IntStream.range(0, 5).mapToObj(i -> "foo-" + i).collect(toList());
+
+    int connectionCount = Host.listConnections().size();
+
     for (String consumerName : consumerNames) {
       Client c0 =
           cf.get(
@@ -417,6 +420,7 @@ void killingConnectionsShouldTriggerConsumerUpdateNotification() throws Exceptio
                         consumerStates.put(consumerName + "-connection-0", active);
                         return null;
                       }));
+      connectionCount++;
       Client c1 =
           cf.get(
               withConnectionName(consumerName + "-connection-1")
@@ -425,6 +429,7 @@ void killingConnectionsShouldTriggerConsumerUpdateNotification() throws Exceptio
                         consumerStates.put(consumerName + "-connection-1", active);
                         return null;
                       }));
+      connectionCount++;
 
       Map<String, String> subscriptionProperties = new HashMap<>();
       subscriptionProperties.put("single-active-consumer", "true");
@@ -440,6 +445,9 @@ void killingConnectionsShouldTriggerConsumerUpdateNotification() throws Exceptio
       assertThat(response).is(ok());
     }
 
+    int cCount = connectionCount;
+    waitAtMost(() -> Host.listConnections().size() == cCount);
+
     for (String consumerName : consumerNames) {
       Host.killConnection(consumerName + "-connection-0");
       waitAtMost(