Repeat test in case of failure

It involves retention, which is flaky on CI.

Author: acogoluegnes

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

@@ -13,11 +13,8 @@
 // [email protected].
 package com.rabbitmq.stream.impl;
 
+import static com.rabbitmq.stream.impl.TestUtils.*;
 import static com.rabbitmq.stream.impl.TestUtils.ResponseConditions.*;
-import static com.rabbitmq.stream.impl.TestUtils.b;
-import static com.rabbitmq.stream.impl.TestUtils.latchAssert;
-import static com.rabbitmq.stream.impl.TestUtils.streamName;
-import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -925,52 +922,44 @@ void streamStatsShouldReturnErrorWhenStreamDoesNotExist() {
 
   @Test
   @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
-  void streamStatsFirstOffsetShouldChangeAfterRetentionKickedIn(TestInfo info) {
+  void streamStatsFirstOffsetShouldChangeAfterRetentionKickedIn(TestInfo info) throws Exception {
     // this test is flaky in some CI environments, so we have to retry it
-    int attemptCount = 0;
-    int maxAttempts = 3;
-    while (attemptCount <= maxAttempts) {
-      attemptCount++;
-      int messageCount = 1000;
-      int payloadSize = 1000;
-      String s = TestUtils.streamName(info);
-      Client client = cf.get();
-      try {
-        assertThat(
-                client
-                    .create(
-                        s,
-                        new Client.StreamParametersBuilder()
-                            .maxLengthBytes(messageCount * payloadSize / 10)
-                            .maxSegmentSizeBytes(messageCount * payloadSize / 20)
-                            .build())
-                    .isOk())
-            .isTrue();
-
-        StreamStatsResponse response = client.streamStats(s);
-        assertThat(response.getInfo()).containsEntry("first_chunk_id", -1L);
-        assertThat(response.getInfo()).containsEntry("committed_chunk_id", -1L);
-
-        byte[] payload = new byte[payloadSize];
-        Function<MessageBuilder, Message> messageCreation = mb -> mb.addData(payload).build();
-
-        TestUtils.publishAndWaitForConfirms(cf, messageCreation, messageCount, s);
-        // publishing again, to make sure new segments trigger retention strategy
-        TestUtils.publishAndWaitForConfirms(cf, messageCreation, messageCount, s);
-        response = client.streamStats(s);
-        assertThat(response.getInfo().get("first_chunk_id")).isPositive();
-        assertThat(response.getInfo().get("committed_chunk_id")).isPositive();
-
-        attemptCount = Integer.MAX_VALUE;
-      } catch (AssertionError e) {
-        // if too many attempts, fail the test, otherwise, try again
-        if (attemptCount > maxAttempts) {
-          throw e;
-        }
-      } finally {
-        assertThat(client.delete(s).isOk()).isTrue();
-      }
-    }
+    repeatIfFailure(
+        () -> {
+          int messageCount = 1000;
+          int payloadSize = 1000;
+          String s = TestUtils.streamName(info);
+          Client client = cf.get();
+          try {
+            assertThat(
+                    client
+                        .create(
+                            s,
+                            new Client.StreamParametersBuilder()
+                                .maxLengthBytes(messageCount * payloadSize / 10)
+                                .maxSegmentSizeBytes(messageCount * payloadSize / 20)
+                                .build())
+                        .isOk())
+                .isTrue();
+
+            StreamStatsResponse response = client.streamStats(s);
+            assertThat(response.getInfo()).containsEntry("first_chunk_id", -1L);
+            assertThat(response.getInfo()).containsEntry("committed_chunk_id", -1L);
+
+            byte[] payload = new byte[payloadSize];
+            Function<MessageBuilder, Message> messageCreation = mb -> mb.addData(payload).build();
+
+            TestUtils.publishAndWaitForConfirms(cf, messageCreation, messageCount, s);
+            // publishing again, to make sure new segments trigger retention strategy
+            TestUtils.publishAndWaitForConfirms(cf, messageCreation, messageCount, s);
+            response = client.streamStats(s);
+            assertThat(response.getInfo().get("first_chunk_id")).isPositive();
+            assertThat(response.getInfo().get("committed_chunk_id")).isPositive();
+
+          } finally {
+            assertThat(client.delete(s).isOk()).isTrue();
+          }
+        });
   }
 
   @Test

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

@@ -419,23 +419,4 @@ private void publishAmqp(Connection c, int messageCount, Supplier<String> filter
       ch.waitForConfirmsOrDie();
     }
   }
-
-  private static void repeatIfFailure(RunnableWithException test) throws Exception {
-    int executionCount = 0;
-    Throwable lastException = null;
-    while (executionCount < 5) {
-      try {
-        test.run();
-        return;
-      } catch (Exception | AssertionError e) {
-        executionCount++;
-        lastException = e;
-      }
-    }
-    if (lastException instanceof Error) {
-      throw new RuntimeException(lastException);
-    } else {
-      throw (Exception) lastException;
-    }
-  }
 }

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

@@ -1066,4 +1066,23 @@ static void waitUntilStable(LongSupplier value) {
       }
     }
   }
+
+  static void repeatIfFailure(RunnableWithException test) throws Exception {
+    int executionCount = 0;
+    Throwable lastException = null;
+    while (executionCount < 5) {
+      try {
+        test.run();
+        return;
+      } catch (Exception | AssertionError e) {
+        executionCount++;
+        lastException = e;
+      }
+    }
+    if (lastException instanceof Error) {
+      throw new RuntimeException(lastException);
+    } else {
+      throw (Exception) lastException;
+    }
+  }
 }