This is an automated email from the ASF dual-hosted git repository.
huaxingao 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 3316c1d1a4 Core: Do not retry Tasks after interrupt (#18213)
3316c1d1a4 is described below
commit 3316c1d1a4bd36712641b98e11d66dee5a74d0d6
Author: Huaxin Gao <[email protected]>
AuthorDate: Thu Sep 24 14:18:24 2026 -0700
Core: Do not retry Tasks after interrupt (#18213)
* Core: Do not retry Tasks after interrupt
* address comments
* Add parallel test for interrupt without retry
* address comments
---
.../main/java/org/apache/iceberg/util/Tasks.java | 28 +++++
.../java/org/apache/iceberg/util/TestTasks.java | 140 +++++++++++++++++++++
2 files changed, 168 insertions(+)
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 29100c6cff..638736c45e 100644
--- a/core/src/main/java/org/apache/iceberg/util/Tasks.java
+++ b/core/src/main/java/org/apache/iceberg/util/Tasks.java
@@ -18,6 +18,9 @@
*/
package org.apache.iceberg.util;
+import java.io.InterruptedIOException;
+import java.net.SocketTimeoutException;
+import java.nio.channels.ClosedByInterruptException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -414,6 +417,13 @@ public class Tasks {
break;
} catch (Exception e) {
+ // a failure caused by cancellation is not transient: retrying would
do real work for a
+ // caller that has already gone away
+ if (causedByInterruption(e)) {
+ Thread.currentThread().interrupt();
+ throw e;
+ }
+
long durationMs = System.currentTimeMillis() - start;
if (attempt >= maxAttempts || (durationMs > maxDurationMs && attempt
> 1)) {
if (durationMs > maxDurationMs) {
@@ -470,6 +480,24 @@ public class Tasks {
}
}
+ private static boolean causedByInterruption(Throwable throwable) {
+ Throwable current = throwable;
+ while (current != null) {
+ // SocketTimeoutException extends InterruptedIOException but signals a
socket timeout rather
+ // than Thread.interrupt(), so it remains retryable
+ if (current instanceof InterruptedException
+ || current instanceof ClosedByInterruptException
+ || (current instanceof InterruptedIOException
+ && !(current instanceof SocketTimeoutException))) {
+ return true;
+ }
+
+ current = current.getCause();
+ }
+
+ return false;
+ }
+
@SuppressWarnings("checkstyle:CyclomaticComplexity")
private static Collection<Throwable> waitFor(Collection<Future<?>> futures) {
while (true) {
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 2ca69c66bb..c46f5a2d6e 100644
--- a/core/src/test/java/org/apache/iceberg/util/TestTasks.java
+++ b/core/src/test/java/org/apache/iceberg/util/TestTasks.java
@@ -19,11 +19,25 @@
package org.apache.iceberg.util;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.io.UncheckedIOException;
+import java.net.SocketTimeoutException;
+import java.nio.channels.ClosedByInterruptException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
+import java.util.stream.Stream;
import org.apache.iceberg.metrics.Counter;
import org.apache.iceberg.metrics.DefaultMetricsContext;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
public class TestTasks {
@@ -57,4 +71,130 @@ public class TestTasks {
assertThat(counter.value()).isOne();
}
+
+ @ParameterizedTest
+ @MethodSource("interruptions")
+ void tasksAreNotRetriedWhenFailureIsCausedByInterruption(Exception
interruption) {
+ Counter counter = new DefaultMetricsContext().counter("counter");
+
+ try {
+ assertThatThrownBy(
+ () ->
+ Tasks.foreach(1)
+ .countAttempts(counter)
+ .retry(3)
+ .run(
+ x -> {
+ throw interruption;
+ },
+ Exception.class))
+ .isSameAs(interruption);
+
+ assertThat(counter.value()).as("Interrupted task should not be
retried").isOne();
+ assertThat(Thread.currentThread().isInterrupted())
+ .as("Interrupt status should be restored")
+ .isTrue();
+ } finally {
+ Thread.interrupted();
+ }
+ }
+
+ private static Stream<Exception> interruptions() {
+ return Stream.of(
+ new InterruptedException("interrupted"),
+ new InterruptedIOException("interrupted"),
+ new ClosedByInterruptException(),
+ new RuntimeException("failed to read manifest", new
InterruptedException("interrupted")),
+ new UncheckedIOException(new InterruptedIOException("interrupted")),
+ new RuntimeException(
+ "failed to read manifest",
+ new IOException("failed to open stream", new
ClosedByInterruptException())));
+ }
+
+ @ParameterizedTest
+ @MethodSource("transientFailures")
+ void tasksAreRetriedWhenFailureIsNotCausedByInterruption(Exception failure) {
+ Counter counter = new DefaultMetricsContext().counter("counter");
+
+ assertThatThrownBy(
+ () ->
+ Tasks.foreach(1)
+ .countAttempts(counter)
+ .exponentialBackoff(0, 0, 5000, 0)
+ .retry(3)
+ .run(
+ x -> {
+ throw failure;
+ },
+ Exception.class))
+ .isSameAs(failure);
+
+ assertThat(counter.value()).isEqualTo(4);
+ assertThat(Thread.currentThread().isInterrupted()).isFalse();
+ }
+
+ private static Stream<Exception> transientFailures() {
+ return Stream.of(
+ new RuntimeException("failed to read manifest", new
IOException("connection reset")),
+ // SocketTimeoutException extends InterruptedIOException but is not an
interrupt
+ new SocketTimeoutException("read timed out"),
+ new RuntimeException("failed to read manifest", new
SocketTimeoutException("timed out")));
+ }
+
+ @Test
+ void parallelTasksAreNotRetriedWhenTheCallerIsInterrupted() throws
InterruptedException {
+ int taskCount = 3;
+ Counter counter = new DefaultMetricsContext().counter("counter");
+ CountDownLatch allTasksBlocked = new CountDownLatch(taskCount);
+ CountDownLatch allTasksInterrupted = new CountDownLatch(taskCount);
+
+ ExecutorService taskPool = Executors.newFixedThreadPool(taskCount);
+ ExecutorService callerPool = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> caller =
+ callerPool.submit(
+ () ->
+ Tasks.range(taskCount)
+ .countAttempts(counter)
+ .retry(3)
+ .suppressFailureWhenFinished()
+ .executeWith(taskPool)
+ .run(
+ index -> {
+ allTasksBlocked.countDown();
+ try {
+ Thread.sleep(TimeUnit.MINUTES.toMillis(10));
+ } catch (InterruptedException e) {
+ // blocking reads surface an interrupt as a
wrapped failure without
+ // preserving the interrupt status
+ allTasksInterrupted.countDown();
+ throw new RuntimeException("failed to read", e);
+ }
+ }));
+
+ assertThat(allTasksBlocked.await(5, TimeUnit.SECONDS)).isTrue();
+
+ // interrupts the caller while it waits in Tasks.waitFor, which cancels
every running task
+ caller.cancel(true);
+
+ assertThat(allTasksInterrupted.await(5, TimeUnit.SECONDS))
+ .as("Each task should be interrupted individually")
+ .isTrue();
+
+ // a retried task never releases its thread: it sleeps for the retry
backoff and then blocks
+ // again, so an idle pool means the interrupted tasks stopped
+ CountDownLatch poolIsIdle = new CountDownLatch(1);
+ taskPool.submit(poolIsIdle::countDown);
+
+ assertThat(poolIsIdle.await(5, TimeUnit.SECONDS))
+ .as("Interrupted tasks should release their threads instead of
retrying")
+ .isTrue();
+ assertThat(counter.value())
+ .as("None of the interrupted tasks should be retried")
+ .isEqualTo(taskCount);
+ } finally {
+ taskPool.shutdownNow();
+ callerPool.shutdownNow();
+ }
+ }
}
