[iceberg] branch master updated: Core: Fix flaky TestParallelIterable test (#7372)

[ja...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jackye pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/master by this push:
     new 5c85dd3459 Core: Fix flaky TestParallelIterable test (#7372)
5c85dd3459 is described below

commit 5c85dd345934f0148bf3c36bb563f9a89c31dfb6
Author: Amogh Jahagirdar <ja...@amazon.com>
AuthorDate: Tue Apr 18 20:45:25 2023 -0700

    Core: Fix flaky TestParallelIterable test (#7372)
---
 .../org/apache/iceberg/util/TestParallelIterable.java  | 18 +++++++++++++++---
 1 file changed, 15 insertions(+), 3 deletions(-)

diff --git a/core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java b/core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java
index 81d06dbc25..960f7d454e 100644
--- a/core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java
+++ b/core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java
@@ -23,13 +23,16 @@ import static org.assertj.core.api.Assertions.assertThat;
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.Collections;
+import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.CloseableIterator;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.awaitility.Awaitility;
 import org.junit.Test;
 
 public class TestParallelIterable {
@@ -60,9 +63,18 @@ public class TestParallelIterable {
 
     assertThat(iterator.hasNext()).isTrue();
     assertThat(iterator.next()).isNotNull();
-    assertThat(queue).isNotEmpty();
-
+    Awaitility.await("Queue is populated")
+        .atMost(5, TimeUnit.SECONDS)
+        .untilAsserted(() -> queueHasElements(iterator, queue));
     iterator.close();
-    assertThat(queue).isEmpty();
+    Awaitility.await("Queue is cleared")
+        .atMost(5, TimeUnit.SECONDS)
+        .untilAsserted(() -> assertThat(queue).isEmpty());
+  }
+
+  private void queueHasElements(CloseableIterator<Integer> iterator, Queue queue) {
+    assertThat(iterator.hasNext()).isTrue();
+    assertThat(iterator.next()).isNotNull();
+    assertThat(queue).isNotEmpty();
   }
 }