This is an automated email from the ASF dual-hosted git repository.
RussellSpitzer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 1fbd13e42b Core: Block on unfinished futures in Tasks.waitFor instead
of polling (#18540)
1fbd13e42b is described below
commit 1fbd13e42b285359f62828ee2743f78537006ba0
Author: Raunaq Morarka <[email protected]>
AuthorDate: Thu Oct 29 22:36:32 2026 +0530
Core: Block on unfinished futures in Tasks.waitFor instead of polling
(#18540)
---
.../main/java/org/apache/iceberg/util/Tasks.java | 90 ++++++-------
.../org/apache/iceberg/TestRemoveSnapshots.java | 12 +-
.../java/org/apache/iceberg/util/TestTasks.java | 140 +++++++++++++++++++++
3 files changed, 184 insertions(+), 58 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/util/Tasks.java
b/core/src/main/java/org/apache/iceberg/util/Tasks.java
index df28528ba3..e167662ae0 100644
--- a/core/src/main/java/org/apache/iceberg/util/Tasks.java
+++ b/core/src/main/java/org/apache/iceberg/util/Tasks.java
@@ -503,67 +503,51 @@ public class Tasks {
return false;
}
- @SuppressWarnings("checkstyle:CyclomaticComplexity")
private static Collection<Throwable> waitFor(Collection<Future<?>> futures) {
- while (true) {
- int numFinished = 0;
+ List<Throwable> uncaught = Lists.newArrayList();
+ Error error = null;
+ try {
+ // block on each future in turn and collect any uncaught exceptions
for (Future<?> future : futures) {
- if (future.isDone()) {
- numFinished += 1;
- }
- }
-
- if (numFinished == futures.size()) {
- List<Throwable> uncaught = Lists.newArrayList();
- // all of the futures are done, get any uncaught exceptions
- for (Future<?> future : futures) {
- try {
- future.get();
-
- } catch (InterruptedException e) {
- LOG.warn("Interrupted while getting future results", e);
- for (Throwable t : uncaught) {
- e.addSuppressed(t);
- }
- Thread.currentThread().interrupt();
- throw new RuntimeException(e);
-
- } catch (CancellationException e) {
- // ignore cancellations
-
- } catch (ExecutionException e) {
- Throwable cause = e.getCause();
- if (cause instanceof Error) {
- for (Throwable t : uncaught) {
- cause.addSuppressed(t);
- }
- throw (Error) cause;
- }
-
- if (cause != null) {
- uncaught.add(e);
- }
-
- LOG.warn("Task threw uncaught exception", cause);
+ try {
+ future.get();
+
+ } catch (CancellationException e) {
+ // ignore cancellations
+
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof Error && error == null) {
+ // thrown once all futures are done
+ error = (Error) cause;
+ } else if (cause != null) {
+ uncaught.add(e);
}
- }
- return uncaught;
+ LOG.warn("Task threw uncaught exception", cause);
+ }
+ }
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted while waiting for tasks to finish", e);
+ for (Throwable t : uncaught) {
+ e.addSuppressed(t);
+ }
- } else {
- try {
- Thread.sleep(10);
- } catch (InterruptedException e) {
- LOG.warn("Interrupted while waiting for tasks to finish", e);
+ for (Future<?> future : futures) {
+ future.cancel(true);
+ }
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
- for (Future<?> future : futures) {
- future.cancel(true);
- }
- Thread.currentThread().interrupt();
- throw new RuntimeException(e);
- }
+ if (error != null) {
+ for (Throwable t : uncaught) {
+ error.addSuppressed(t);
}
+ throw error;
}
+
+ return uncaught;
}
/** A range, [ 0, size ) */
diff --git a/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java
b/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java
index e3dd980dfe..0348972241 100644
--- a/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java
+++ b/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java
@@ -1671,7 +1671,9 @@ public class TestRemoveSnapshots extends TestBase {
long lastSnapshotId = tableWithBulkIO.currentSnapshot().snapshotId();
-
removeSnapshots(tableWithBulkIO).expireOlderThan(System.currentTimeMillis()).commit();
+ removeSnapshots(tableWithBulkIO)
+ .expireOlderThan(tableWithBulkIO.currentSnapshot().timestampMillis() +
1)
+ .commit();
assertThat(tableWithBulkIO.currentSnapshot().snapshotId()).isEqualTo(lastSnapshotId);
assertThat(tableWithBulkIO.snapshots()).containsOnly(tableWithBulkIO.currentSnapshot());
@@ -1720,7 +1722,7 @@ public class TestRemoveSnapshots extends TestBase {
Set<String> deletedFiles = Sets.newHashSet();
// Expiring snapshots should remove the data_bucket partition
removeSnapshots(table)
- .expireOlderThan(System.currentTimeMillis())
+ .expireOlderThan(table.currentSnapshot().timestampMillis() + 1)
.cleanExpiredMetadata(true)
.deleteWith(deletedFiles::add)
.commit();
@@ -1755,7 +1757,7 @@ public class TestRemoveSnapshots extends TestBase {
Set<String> deletedFiles = Sets.newHashSet();
// Expiring snapshots should remove only the unpartitioned spec
removeSnapshots(table)
- .expireOlderThan(System.currentTimeMillis())
+ .expireOlderThan(table.currentSnapshot().timestampMillis() + 1)
.cleanExpiredMetadata(true)
.deleteWith(deletedFiles::add)
.commit();
@@ -1787,7 +1789,7 @@ public class TestRemoveSnapshots extends TestBase {
Set<String> deletedFiles = Sets.newHashSet();
// Expire all snapshots and schemas except the current ones.
removeSnapshots(table)
- .expireOlderThan(System.currentTimeMillis())
+ .expireOlderThan(table.currentSnapshot().timestampMillis() + 1)
.cleanExpiredMetadata(true)
.deleteWith(deletedFiles::add)
.commit();
@@ -1821,7 +1823,7 @@ public class TestRemoveSnapshots extends TestBase {
Set<String> deletedFiles = Sets.newHashSet();
// Expire all snapshots except the current one. No unused schemas or specs
to be removed.
removeSnapshots(table)
- .expireOlderThan(System.currentTimeMillis())
+ .expireOlderThan(table.currentSnapshot().timestampMillis() + 1)
.cleanExpiredMetadata(true)
.deleteWith(deletedFiles::add)
.commit();
diff --git a/core/src/test/java/org/apache/iceberg/util/TestTasks.java
b/core/src/test/java/org/apache/iceberg/util/TestTasks.java
index c46f5a2d6e..3263b776a7 100644
--- a/core/src/test/java/org/apache/iceberg/util/TestTasks.java
+++ b/core/src/test/java/org/apache/iceberg/util/TestTasks.java
@@ -26,21 +26,44 @@ import java.io.InterruptedIOException;
import java.io.UncheckedIOException;
import java.net.SocketTimeoutException;
import java.nio.channels.ClosedByInterruptException;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RunnableFuture;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import org.apache.iceberg.metrics.Counter;
import org.apache.iceberg.metrics.DefaultMetricsContext;
+import
org.apache.iceberg.relocated.com.google.common.util.concurrent.Uninterruptibles;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
public class TestTasks {
+ private RecordingExecutor executor;
+
+ @BeforeEach
+ void before() {
+ executor = new RecordingExecutor(2);
+ }
+
+ @AfterEach
+ void after() {
+ executor.shutdownNow();
+ }
+
@Test
public void attemptCounterIsIncreasedOnRetries() {
Counter counter = new DefaultMetricsContext().counter("counter");
@@ -197,4 +220,121 @@ public class TestTasks {
callerPool.shutdownNow();
}
}
+
+ @Test
+ 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
+ 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
+ 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")
+ .hasCauseInstanceOf(InterruptedException.class);
+
+ assertThat(Thread.interrupted()).isTrue();
+ assertThat(executor.futures()).hasSize(4).allMatch(Future::isCancelled);
+ }
+
+ private static class RecordingExecutor extends ThreadPoolExecutor {
+ private final List<RecordingFuture<?>> futures = new
CopyOnWriteArrayList<>();
+
+ RecordingExecutor(int threads) {
+ super(threads, threads, 0L, TimeUnit.MILLISECONDS, new
LinkedBlockingQueue<>());
+ }
+
+ List<RecordingFuture<?>> futures() {
+ return futures;
+ }
+
+ void awaitBlockedGet(int index) {
+ futures.get(index).awaitBlockedGet();
+ }
+
+ @Override
+ protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
+ RecordingFuture<T> future = new RecordingFuture<>(runnable, value);
+ futures.add(future);
+ return future;
+ }
+ }
+
+ private static class RecordingFuture<T> extends FutureTask<T> {
+ // counts down when get() is called before the task is done
+ private final CountDownLatch blockedGet = new CountDownLatch(1);
+
+ RecordingFuture(Runnable runnable, T value) {
+ super(runnable, value);
+ }
+
+ void awaitBlockedGet() {
+ assertThat(Uninterruptibles.awaitUninterruptibly(blockedGet, 10,
TimeUnit.SECONDS))
+ .as("get() should be called before the task is done")
+ .isTrue();
+ }
+
+ @Override
+ public T get() throws InterruptedException, ExecutionException {
+ if (!isDone()) {
+ blockedGet.countDown();
+ }
+ return super.get();
+ }
+ }
}
