Attempt to fix flaky test by not requiring advance to succeed immediately

Author: scwhittle

sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java

@@ -25,6 +25,7 @@
 import static org.apache.beam.sdk.io.jms.CommonJms.toSerializableFunction;
 import static org.apache.beam.sdk.io.jms.JmsIO.Writer.JMS_IO_PRODUCER_METRIC_NAME;
 import static org.apache.beam.sdk.io.jms.JmsIO.Writer.PUBLICATION_RETRIES_METRIC_NAME;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
 import static org.hamcrest.CoreMatchers.allOf;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.contains;
@@ -558,7 +559,15 @@ public void testCheckpointMark() throws Exception {
 
     // consume 3 messages (NB: start already consumed the first message)
     for (int i = 0; i < 3; i++) {
-      assertTrue(String.format("Failed at %d-th message", i), reader.advance());
+      boolean readerAdvanced = false;
+      for (int attempt = 0; attempt < 10; attempt++) {
+        if (reader.advance()) {
+          readerAdvanced = true;
+          break;
+        }
+        sleepUninterruptibly(java.time.Duration.ofMillis(100));
+      }
+      assertTrue(String.format("Failed at %d-th message", i), readerAdvanced);
     }
 
     // the messages are still pending in the queue (no ACK yet)