diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/retrying/DirectRetryingExecutor.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/retrying/DirectRetryingExecutor.java index aeff549e7542..bdcc4b03a2fd 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/retrying/DirectRetryingExecutor.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/retrying/DirectRetryingExecutor.java @@ -34,6 +34,7 @@ import com.google.api.core.ApiFuture; import com.google.api.core.ApiFutures; import java.io.InterruptedIOException; +import java.net.SocketTimeoutException; import java.nio.channels.ClosedByInterruptException; import java.util.concurrent.Callable; import org.jspecify.annotations.NullMarked; @@ -103,6 +104,11 @@ public ApiFuture submit(RetryingFuture retryingFuture) { sleep(retryingFuture.getAttemptSettings().getRandomizedRetryDelayDuration()); ResponseT response = retryingFuture.getCallable().call(); retryingFuture.setAttemptFuture(ApiFutures.immediateFuture(response)); + } catch (SocketTimeoutException e) { + // A connect or read timeout is an InterruptedIOException, but no thread was interrupted. + // Setting the interrupt flag here makes setAttemptFuture throw a new InterruptedException, + // which drops this exception and stops the retry. Let the retry algorithm judge it. + retryingFuture.setAttemptFuture(ApiFutures.immediateFailedFuture(e)); } catch (InterruptedException | InterruptedIOException | ClosedByInterruptException e) { Thread.currentThread().interrupt(); retryingFuture.setAttemptFuture(ApiFutures.immediateFailedFuture(e)); diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/retrying/DirectRetryingExecutorTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/retrying/DirectRetryingExecutorTest.java index 28c0154e911f..2b60fe274ec3 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/retrying/DirectRetryingExecutorTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/retrying/DirectRetryingExecutorTest.java @@ -29,7 +29,19 @@ */ package com.google.api.gax.retrying; +import static com.google.api.gax.retrying.FailingCallable.FAST_RETRY_SETTINGS; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; + import com.google.api.core.CurrentMillisClock; +import java.io.InterruptedIOException; +import java.net.SocketTimeoutException; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; class DirectRetryingExecutorTest extends AbstractRetryingExecutorTest { @@ -45,4 +57,61 @@ protected RetryAlgorithm getAlgorithm( new TestResultRetryAlgorithm(apocalypseCountDown, apocalypseException), new ExponentialRetryAlgorithm(retrySettings, CurrentMillisClock.getDefaultClock())); } + + /** + * Runs {@code callable} under an algorithm that retries a {@link SocketTimeoutException} and + * nothing else, so the test sees which exception the algorithm was given. + */ + private RetryingFuture runRetryingTimeouts(Callable callable) { + setUp(false); + RetryAlgorithm algorithm = + new RetryAlgorithm<>( + new BasicResultRetryAlgorithm() { + @Override + public boolean shouldRetry(Throwable prevThrowable, String prevResponse) { + return prevThrowable instanceof SocketTimeoutException; + } + }, + new ExponentialRetryAlgorithm( + FAST_RETRY_SETTINGS, CurrentMillisClock.getDefaultClock())); + RetryingExecutorWithContext executor = getExecutor(algorithm); + RetryingFuture future = executor.createFuture(callable, retryingContext); + future.setAttemptFuture(executor.submit(future)); + return future; + } + + @Test + void testSocketTimeoutReachesTheRetryAlgorithm() throws Exception { + AtomicInteger calls = new AtomicInteger(); + try { + RetryingFuture future = + runRetryingTimeouts( + () -> { + if (calls.getAndIncrement() == 0) { + throw new SocketTimeoutException("Read timed out"); + } + return "SUCCESS"; + }); + assertFalse(Thread.currentThread().isInterrupted()); + assertEquals("SUCCESS", future.get()); + assertEquals(2, calls.get()); + } finally { + Thread.interrupted(); + } + } + + @Test + void testInterruptedIOExceptionStillFailsAsInterrupted() { + try { + RetryingFuture future = + runRetryingTimeouts( + () -> { + throw new InterruptedIOException("interrupted"); + }); + ExecutionException e = assertThrows(ExecutionException.class, future::get); + assertInstanceOf(InterruptedException.class, e.getCause()); + } finally { + Thread.interrupted(); + } + } }