raunaqmorarka commented on code in PR #18540:
URL: https://github.com/apache/iceberg/pull/18540#discussion_r4405963508
##########
core/src/test/java/org/apache/iceberg/util/TestTasks.java:
##########
@@ -197,4 +222,143 @@ void
parallelTasksAreNotRetriedWhenTheCallerIsInterrupted() throws InterruptedEx
callerPool.shutdownNow();
}
}
+
+ @Test
+ public void parallelRunReturnsWhenTasksFinish() {
+ int rounds = 100;
+ AtomicInteger finished = new AtomicInteger();
+
+ long start = System.nanoTime();
+ for (int round = 0; round < rounds; round++) {
+ Tasks.range(4)
+ .executeWith(executor)
+ .run(
+ x -> {
+ LockSupport.parkNanos(TimeUnit.MICROSECONDS.toNanos(100));
+ finished.incrementAndGet();
+ });
+ assertThat(finished).hasValue((round + 1) * 4);
+ }
+ long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
start);
+
+ // polling with a 10 ms sleep takes at least 10 ms per round
+ assertThat(elapsedMillis).isLessThan(rounds * 10L);
+ }
+
+ @Test
+ public void parallelRunRethrowsTaskErrorAfterOtherTasksFinish() {
+ AtomicBoolean otherTaskFinished = new AtomicBoolean();
+
+ assertThatThrownBy(
+ () ->
+ Tasks.range(2)
+ .executeWith(executor)
+ .run(
+ x -> {
+ if (x == 0) {
+ throw new Error("task failed");
+ }
+ executor.awaitBlockedGet(1);
+ otherTaskFinished.set(true);
+ }))
+ .isExactlyInstanceOf(Error.class)
+ .hasMessage("task failed");
+
+ assertThat(otherTaskFinished).isTrue();
+ }
+
+ @Test
+ public void parallelRunIgnoresCancelledFuture() throws InterruptedException {
+ CountDownLatch release = new CountDownLatch(1);
+
+ boolean succeeded =
+ Tasks.range(2)
+ .executeWith(executor)
+ .run(
+ x -> {
+ if (x == 0) {
+ release.await(10, TimeUnit.SECONDS);
+ } else {
+ executor.awaitBlockedGet(0);
+ executor.futures().get(0).cancel(false);
+ release.countDown();
+ }
+ },
+ InterruptedException.class);
+
+ assertThat(succeeded).isTrue();
+ assertThat(executor.futures().get(0).isCancelled()).isTrue();
+ }
+
+ @Test
+ public void parallelRunCancelsFuturesOnInterrupt() {
+ Thread waitingThread = Thread.currentThread();
+ CountDownLatch neverReleased = new CountDownLatch(1);
+
+ assertThatThrownBy(
+ () ->
+ Tasks.range(4)
+ .executeWith(executor)
+ .run(
+ x -> {
+ if (x == 0) {
+ executor.awaitBlockedGet(0);
+ waitingThread.interrupt();
+ }
+ neverReleased.await(10, TimeUnit.SECONDS);
+ },
+ InterruptedException.class))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("InterruptedException")
Review Comment:
Dropping it fails checkstyle. `AssertThatThrownByWithMessageCheck` requires
a message check on every `assertThatThrownBy`.
The message here is just the cause's `toString()`, so this is the closest
check we have.
--
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]
