Log more info in cluster test

rabbitmq · Dec 19, 2024 · 2c219c6 · 2c219c6
1 parent 979526b
commit 2c219c6
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 3 deletions.
diff --git a/src/main/java/com/rabbitmq/stream/impl/ProducersCoordinator.java b/src/main/java/com/rabbitmq/stream/impl/ProducersCoordinator.java
@@ -750,7 +750,7 @@ private void assignProducersToNewManagers(
                 List<BrokerWrapper> candidates = brokerAndCandidates.v2();
                 String key = keyForNode(broker);
                 LOGGER.debug(
-                    "Assigning {} producer(s) and consumer tracker(s) to {}", trackers.size(), key);
+                    "Assigning {} producer(s) and consumer tracker(s) to {} (stream '{}')", trackers.size(), key, stream);
                 trackers.forEach(tracker -> maybeRecoverAgent(broker, candidates, tracker));
               })
           .exceptionally(

diff --git a/src/test/java/com/rabbitmq/stream/impl/RecoveryClusterTest.java b/src/test/java/com/rabbitmq/stream/impl/RecoveryClusterTest.java
@@ -34,6 +34,9 @@
 import io.netty.channel.EventLoopGroup;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -251,6 +254,11 @@ void clusterRestart(boolean useLoadBalancer, boolean forceLeader) throws Interru
       LOGGER.info("Environment information:");
       System.out.println(TestUtils.jsonPrettyPrint(environment.toString()));
 
+      LOGGER.info("Producer information:");
+      producers.forEach(p -> {
+        LOGGER.info("Producer to '{}' (last exception: '{}')", p.stream(), p.lastException);
+      });
+
       LOGGER.info("Closing producers");
       producers.forEach(
           p -> {
@@ -293,6 +301,8 @@ private static class ProducerState implements AutoCloseable {
     final AtomicBoolean stopped = new AtomicBoolean(false);
     final AtomicInteger acceptedCount = new AtomicInteger();
     final AtomicReference<Runnable> postConfirmed = new AtomicReference<>(() -> {});
+    final AtomicReference<Throwable> lastException = new AtomicReference<>();
+    final AtomicReference<Instant> lastExceptionInstant = new AtomicReference<>();
 
     private ProducerState(String stream, boolean dynamicBatch, Environment environment) {
       this.stream = stream;
@@ -317,8 +327,9 @@ void start() {
                         this.limiter.acquire(1);
                         this.producer.send(
                             producer.messageBuilder().addData(BODY).build(), confirmationHandler);
-                      } catch (Exception e) {
-
+                      } catch (Throwable e) {
+                        this.lastException.set(e);
+                        this.lastExceptionInstant.set(Instant.now());
                       }
                     }
                   });
@@ -342,6 +353,14 @@ String stream() {
       return this.stream;
     }
 
+    String lastException() {
+      if (this.lastException.get() == null) {
+        return "no exception";
+      } else {
+        return this.lastException.get().getMessage() + " at " + DateTimeFormatter.ISO_LOCAL_TIME.format(lastExceptionInstant.get());
+      }
+    }
+
     @Override
     public void close() {
       stopped.set(true);