raunaqmorarka commented on code in PR #18522:
URL: https://github.com/apache/iceberg/pull/18522#discussion_r4377304183
##########
core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java:
##########
@@ -68,11 +73,11 @@ public CloseableIterator<Integer> iterator() {
assertThat(iterator.hasNext()).isTrue();
assertThat(iterator.next()).isNotNull();
Awaitility.await("Queue is populated")
- .atMost(5, TimeUnit.SECONDS)
+ .atMost(10, TimeUnit.SECONDS)
.untilAsserted(() -> queueHasElements(iterator));
iterator.close();
Awaitility.await("Queue is cleared")
- .atMost(5, TimeUnit.SECONDS)
+ .atMost(10, TimeUnit.SECONDS)
Review Comment:
Why do these waits, and the two at lines 136-140, go from 5 to 10 seconds?
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -208,14 +227,18 @@ public synchronized boolean hasNext() {
// this cannot conclude that there are no more records until tasks have
finished. while some
// are running, return true when there is at least one item to return.
- while (checkTasks()) {
+ while (true) {
+ taskSignal.drainPermits();
Review Comment:
Please pin this drain above `checkTasks()` with a test and a one-line
comment, since moving it below lets `acquire()` block forever while the current
tests pass. A task that finishes before `whenComplete` registers releases its
permit inside `checkTasks()`, and a later drain discards it. This test hangs
with the drain moved just above `acquire()`:
```java
@Test
@Timeout(value = 30, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
void wakesWhenTaskCompletesInline() throws Exception {
try (ParallelIterable<Integer> iterable =
new ParallelIterable<>(List.of(List.of()),
MoreExecutors.newDirectExecutorService())) {
assertThat(iterable.iterator().hasNext()).isFalse();
}
}
```
##########
core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java:
##########
@@ -213,6 +218,149 @@ public void noDeadlock() {
}
}
+ @Test
+ @Timeout(60)
+ void wakesOnDataBeforeTaskCompletes() throws Exception {
+ Semaphore producerGate = new Semaphore(0);
+ Semaphore completionGate = new Semaphore(0);
+ Iterable<Integer> input =
+ () ->
+ new Iterator<>() {
+ private boolean produced;
+
+ @Override
+ public boolean hasNext() {
+ if (produced) {
+ completionGate.acquireUninterruptibly();
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public Integer next() {
+ producerGate.acquireUninterruptibly();
+ produced = true;
+ return 1;
+ }
+ };
+
+ int result =
+ consumeAfterWakeUp(
+ input,
+ CloseableIterator::next,
+ ignored -> producerGate.release(),
+ producerGate,
+ completionGate);
+ assertThat(result).isEqualTo(1);
+ }
+
+ @Test
+ @Timeout(60)
+ void wakesWhenEmptyTaskCompletes() throws Exception {
+ Semaphore producerGate = new Semaphore(0);
+ Iterable<Integer> input =
+ () -> {
+ producerGate.acquireUninterruptibly();
+ return Collections.emptyIterator();
+ };
+
+ boolean result =
+ consumeAfterWakeUp(
+ input, CloseableIterator::hasNext, ignored ->
producerGate.release(), producerGate);
+ assertThat(result).isFalse();
+ }
+
+ @Test
+ @Timeout(60)
+ void wakesWhenTaskFails() {
+ Semaphore producerGate = new Semaphore(0);
+ IllegalArgumentException failure = new IllegalArgumentException("Cannot
read manifest");
+ Iterable<Integer> input =
+ () -> {
+ producerGate.acquireUninterruptibly();
+ throw failure;
+ };
+
+ assertThatThrownBy(
+ () ->
+ consumeAfterWakeUp(
+ input,
+ CloseableIterator::hasNext,
+ ignored -> producerGate.release(),
+ producerGate))
+ .isInstanceOf(ExecutionException.class)
+ .hasMessage(failure.toString())
+ .hasCause(failure);
+ }
+
+ @Test
+ @Timeout(60)
Review Comment:
Run this test on its own thread, so a lost wake-up in `close()` fails it
after 60 seconds instead of hanging the build. With that wake-up removed,
`close()` waits to enter `synchronized (this)` while the consumer holds the
monitor in `hasNext()`. That wait ignores the interrupt the default timeout
mode sends to the test thread.
```suggestion
@Timeout(value = 60, threadMode = Timeout.ThreadMode.SEPARATE_THREAD)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
