Add logging for a flaky test

Fails only on CI, could not reproduce locally.

Author: acogoluegnes

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

@@ -19,6 +19,7 @@
 import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
 import static org.assertj.core.api.Assertions.assertThat;
 
+import ch.qos.logback.classic.Level;
 import com.rabbitmq.stream.BackOffDelayPolicy;
 import com.rabbitmq.stream.ConfirmationHandler;
 import com.rabbitmq.stream.ConfirmationStatus;
@@ -333,58 +334,66 @@ void shouldRecoverAfterConnectionIsKilled(int subEntrySize) throws Exception {
   @ParameterizedTest
   @ValueSource(ints = {1, 7})
   void producerShouldBeClosedWhenStreamIsDeleted(int subEntrySize, TestInfo info) throws Exception {
-    String s = streamName(info);
-    environment.streamCreator().stream(s).create();
-
-    StreamProducer producer =
-        (StreamProducer) environment.producerBuilder().subEntrySize(subEntrySize).stream(s).build();
-
-    AtomicInteger published = new AtomicInteger(0);
-    AtomicInteger confirmed = new AtomicInteger(0);
-    AtomicInteger errored = new AtomicInteger(0);
-    Set<Number> errorCodes = ConcurrentHashMap.newKeySet();
-
-    short lastExpectedErrorCode = Constants.RESPONSE_CODE_STREAM_DOES_NOT_EXIST;
-    AtomicBoolean continuePublishing = new AtomicBoolean(true);
-    Thread publishThread =
-        new Thread(
-            () -> {
-              ConfirmationHandler confirmationHandler =
-                  confirmationStatus -> {
-                    if (confirmationStatus.isConfirmed()) {
-                      confirmed.incrementAndGet();
-                    } else {
-                      errored.incrementAndGet();
-                      errorCodes.add(confirmationStatus.getCode());
-                      if (confirmationStatus.getCode() == lastExpectedErrorCode) {
-                        continuePublishing.set(false);
+    Level initialLogLevel = TestUtils.newLoggerLevel(ProducersCoordinator.class, Level.DEBUG);
+    try {
+      String s = streamName(info);
+      environment.streamCreator().stream(s).create();
+
+      StreamProducer producer =
+          (StreamProducer)
+              environment.producerBuilder().subEntrySize(subEntrySize).stream(s).build();
+
+      AtomicInteger published = new AtomicInteger(0);
+      AtomicInteger confirmed = new AtomicInteger(0);
+      AtomicInteger errored = new AtomicInteger(0);
+      Set<Number> errorCodes = ConcurrentHashMap.newKeySet();
+
+      short lastExpectedErrorCode = Constants.RESPONSE_CODE_STREAM_DOES_NOT_EXIST;
+      AtomicBoolean continuePublishing = new AtomicBoolean(true);
+      Thread publishThread =
+          new Thread(
+              () -> {
+                ConfirmationHandler confirmationHandler =
+                    confirmationStatus -> {
+                      if (confirmationStatus.isConfirmed()) {
+                        confirmed.incrementAndGet();
+                      } else {
+                        errored.incrementAndGet();
+                        errorCodes.add(confirmationStatus.getCode());
+                        if (confirmationStatus.getCode() == lastExpectedErrorCode) {
+                          continuePublishing.set(false);
+                        }
                       }
-                    }
-                  };
-              while (continuePublishing.get()) {
-                try {
-                  producer.send(
-                      producer
-                          .messageBuilder()
-                          .addData("".getBytes(StandardCharsets.UTF_8))
-                          .build(),
-                      confirmationHandler);
-                  published.incrementAndGet();
-                } catch (StreamException e) {
-                  // OK
+                    };
+                while (continuePublishing.get()) {
+                  try {
+                    producer.send(
+                        producer
+                            .messageBuilder()
+                            .addData("".getBytes(StandardCharsets.UTF_8))
+                            .build(),
+                        confirmationHandler);
+                    published.incrementAndGet();
+                  } catch (StreamException e) {
+                    // OK
+                  }
                 }
-              }
-            });
-    publishThread.start();
+              });
+      publishThread.start();
 
-    Thread.sleep(1000L);
+      Thread.sleep(1000L);
 
-    assertThat(producer.isOpen()).isTrue();
+      assertThat(producer.isOpen()).isTrue();
 
-    environment.deleteStream(s);
+      environment.deleteStream(s);
 
-    waitAtMost(() -> !producer.isOpen());
-    waitAtMost(() -> errorCodes.contains(lastExpectedErrorCode));
+      waitAtMost(() -> !producer.isOpen());
+      waitAtMost(
+          () -> errorCodes.contains(lastExpectedErrorCode),
+          () -> errorCodes + " should contain " + lastExpectedErrorCode);
+    } finally {
+      TestUtils.newLoggerLevel(ProducersCoordinator.class, initialLogLevel);
+    }
   }
 
   @ParameterizedTest

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

@@ -17,6 +17,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.fail;
 
+import ch.qos.logback.classic.Level;
 import com.rabbitmq.client.BuiltinExchangeType;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
@@ -94,6 +95,11 @@ public static Duration waitAtMost(CallableBooleanSupplier condition) throws Exce
     return waitAtMost(10, condition, null);
   }
 
+  public static Duration waitAtMost(CallableBooleanSupplier condition, Supplier<String> message)
+      throws Exception {
+    return waitAtMost(10, condition, message);
+  }
+
   public static Duration waitAtMost(Duration timeout, CallableBooleanSupplier condition)
       throws Exception {
     return waitAtMost((int) timeout.getSeconds(), condition, null);
@@ -799,4 +805,12 @@ void doesNotComplete(Duration timeout) {
       }
     }
   }
+
+  static Level newLoggerLevel(Class<?> c, Level level) {
+    ch.qos.logback.classic.Logger logger =
+        (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(c);
+    Level initialLevel = logger.getEffectiveLevel();
+    logger.setLevel(level);
+    return initialLevel;
+  }
 }