Add polling assertion

Test fails sometimes on CI.

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/amqp/package-info.java

@@ -1,4 +1,2 @@
-/**
- * Classes for AMQP 1.0 support.
- */
-package com.rabbitmq.stream.amqp;
+/** Classes for AMQP 1.0 support. */
+package com.rabbitmq.stream.amqp;

src/main/java/com/rabbitmq/stream/impl/package-info.java

@@ -1,4 +1,4 @@
 /**
  * Implementation-specific, non-public interfaces and classes of the RabbitMQ Stream Java Client.
  */
-package com.rabbitmq.stream.impl;
+package com.rabbitmq.stream.impl;

src/main/java/com/rabbitmq/stream/package-info.java

@@ -1,4 +1,2 @@
-/**
- * Main API for RabbitMQ Stream Java Client.
- */
-package com.rabbitmq.stream;
+/** Main API for RabbitMQ Stream Java Client. */
+package com.rabbitmq.stream;

src/main/java/com/rabbitmq/stream/perf/package-info.java

@@ -1,4 +1,2 @@
-/**
- * Performance tool for RabbitMQ Stream.
- */
-package com.rabbitmq.stream.perf;
+/** Performance tool for RabbitMQ Stream. */
+package com.rabbitmq.stream.perf;

src/main/java/com/rabbitmq/stream/sasl/package-info.java

@@ -1,6 +1,7 @@
 /**
  * SASL API for the RabbitMQ Stream Java Client.
  *
- * @see <a href="https://en.wikipedia.org/wiki/Simple_Authentication_and_Security_Layer">SASL on Wikipedia</a>
+ * @see <a href="https://en.wikipedia.org/wiki/Simple_Authentication_and_Security_Layer">SASL on
+ *     Wikipedia</a>
  */
-package com.rabbitmq.stream.sasl;
+package com.rabbitmq.stream.sasl;

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

@@ -13,6 +13,7 @@
 // [email protected].
 package com.rabbitmq.stream.impl;
 
+import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
@@ -143,8 +144,9 @@ void tearDown() throws Exception {
     "0,1000,7",
   })
   void confirmTimeoutTaskShouldFailMessagesAfterTimeout(
-      long confirmTimeoutMs, long waitTime, int subEntrySize) throws Exception {
+      long confirmTimeoutMs, long waitTimeMs, int subEntrySize) throws Exception {
     Duration confirmTimeout = Duration.ofMillis(confirmTimeoutMs);
+    Duration waitTime = Duration.ofMillis(waitTimeMs);
     clock.refresh();
     int messageCount = 500;
     int confirmedPart = messageCount / 10;
@@ -189,13 +191,14 @@ void confirmTimeoutTaskShouldFailMessagesAfterTimeout(
 
     executorService.scheduleAtFixedRate(() -> clock.refresh(), 100, 100, TimeUnit.MILLISECONDS);
 
-    Thread.sleep(waitTime);
+    Thread.sleep(waitTime.toMillis());
     assertThat(confirmedCount.get()).isEqualTo(expectedConfirmed);
     if (confirmTimeout.isZero()) {
       assertThat(erroredCount.get()).isZero();
       assertThat(responseCodes).isEmpty();
     } else {
-      assertThat(erroredCount.get()).isEqualTo(messageCount - expectedConfirmed);
+      waitAtMost(
+          waitTime.multipliedBy(2), () -> erroredCount.get() == (messageCount - expectedConfirmed));
       assertThat(responseCodes).hasSize(1).contains(Constants.CODE_PUBLISH_CONFIRM_TIMEOUT);
     }
   }