fix(gax): do not treat a SocketTimeoutException as a thread interrupt

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<ResponseT> submit(RetryingFuture<ResponseT> 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.<ResponseT>immediateFailedFuture(e));
       } catch (InterruptedException | InterruptedIOException | ClosedByInterruptException e) {
         Thread.currentThread().interrupt();
         retryingFuture.setAttemptFuture(ApiFutures.<ResponseT>immediateFailedFuture(e));

sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/retrying/DirectRetryingExecutorTest.java

@@ -29,9 +29,21 @@
  */
 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.ApiClock;
 import com.google.api.gax.core.FakeApiClock;
+import java.io.InterruptedIOException;
+import java.net.SocketTimeoutException;
 import java.time.Duration;
+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 {
 
@@ -66,4 +78,60 @@ protected RetryAlgorithm<String> getAlgorithm(
         new TestResultRetryAlgorithm<String>(apocalypseCountDown, apocalypseException),
         new ExponentialRetryAlgorithm(retrySettings, clock));
   }
+
+  /**
+   * 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<String> runRetryingTimeouts(Callable<String> callable) {
+    setUp(false);
+    RetryAlgorithm<String> algorithm =
+        new RetryAlgorithm<>(
+            new BasicResultRetryAlgorithm<String>() {
+              @Override
+              public boolean shouldRetry(Throwable prevThrowable, String prevResponse) {
+                return prevThrowable instanceof SocketTimeoutException;
+              }
+            },
+            new ExponentialRetryAlgorithm(FAST_RETRY_SETTINGS, clock));
+    RetryingExecutorWithContext<String> executor = getExecutor(algorithm);
+    RetryingFuture<String> future = executor.createFuture(callable, retryingContext);
+    future.setAttemptFuture(executor.submit(future));
+    return future;
+  }
+
+  @Test
+  void testSocketTimeoutReachesTheRetryAlgorithm() throws Exception {
+    AtomicInteger calls = new AtomicInteger();
+    try {
+      RetryingFuture<String> 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<String> future =
+          runRetryingTimeouts(
+              () -> {
+                throw new InterruptedIOException("interrupted");
+              });
+      ExecutionException e = assertThrows(ExecutionException.class, future::get);
+      assertInstanceOf(InterruptedException.class, e.getCause());
+    } finally {
+      Thread.interrupted();
+    }
+  }
 }