Fix test

Author: acogoluegnes

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

@@ -260,10 +260,7 @@ public static void setEnv(String parameter, String value) {
   }
 
   public static String rabbitmqctlCommand() {
-    String rabbitmqCtl = System.getProperty("rabbitmqctl.bin");
-    if (rabbitmqCtl == null) {
-      rabbitmqCtl = DOCKER_PREFIX + "rabbitmq";
-    }
+    String rabbitmqCtl = rabbitmqctlBin();
     if (rabbitmqCtl.startsWith(DOCKER_PREFIX)) {
       String containerId = rabbitmqCtl.split(":")[1];
       return "docker exec " + containerId + " rabbitmqctl";
@@ -272,6 +269,14 @@ public static String rabbitmqctlCommand() {
     }
   }
 
+  private static String rabbitmqctlBin() {
+    String rabbitmqCtl = System.getProperty("rabbitmqctl.bin");
+    if (rabbitmqCtl == null) {
+      rabbitmqCtl = DOCKER_PREFIX + "rabbitmq";
+    }
+    return rabbitmqCtl;
+  }
+
   private static String rabbitmqStreamsCommand() {
     String rabbitmqctl = rabbitmqctlCommand();
     int lastIndex = rabbitmqctl.lastIndexOf("rabbitmqctl");
@@ -330,11 +335,7 @@ private static void clearResourceAlarm(String source) {
   }
 
   public static boolean isOnDocker() {
-    String rabbitmqCtl = System.getProperty("rabbitmqctl.bin");
-    if (rabbitmqCtl == null) {
-      throw new IllegalStateException("Please define the rabbitmqctl.bin system property");
-    }
-    return rabbitmqCtl.startsWith(DOCKER_PREFIX);
+    return rabbitmqctlBin().startsWith(DOCKER_PREFIX);
   }
 
   public static List<String> nodes() {

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

@@ -53,10 +53,14 @@
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @ExtendWith(TestUtils.StreamTestInfrastructureExtension.class)
 public class StreamConsumerTest {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(StreamConsumerTest.class);
+
   static final Duration RECOVERY_DELAY = Duration.ofSeconds(2);
   static final Duration TOPOLOGY_DELAY = Duration.ofSeconds(2);
   static volatile Duration recoveryInitialDelay;
@@ -1105,7 +1109,15 @@ private DispatchingMessageHandler(MessageHandler delegate, ThreadFactory tf) {
 
     @Override
     public void handle(Context context, Message message) {
-      this.queue.add(new ContextMessageWrapper(context, message));
+      try {
+        boolean inserted =
+            this.queue.offer(new ContextMessageWrapper(context, message), 10, TimeUnit.SECONDS);
+        if (!inserted) {
+          LOGGER.warn("Failed to insert message into queue");
+        }
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+      }
     }
 
     @Override