Retry flaky retention test

Author: acogoluegnes

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

@@ -932,38 +932,50 @@ void streamStatsShouldReturnErrorWhenStreamDoesNotExist() {
   @Test
   @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
   void streamStatsFirstOffsetShouldChangeAfterRetentionKickedIn(TestInfo info) {
-    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();
+    // 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();
+      }
     }
   }
 }