fix(gax): adjust RetryAlgorithm logic for timeout v. exception

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/retrying/RetryAlgorithm.java

@@ -232,8 +232,18 @@ public boolean shouldRetry(
       ResponseT previousResponse,
       TimedAttemptSettings nextAttemptSettings)
       throws CancellationException {
-    return shouldRetryBasedOnResult(context, previousThrowable, previousResponse)
-        && shouldRetryBasedOnTiming(context, nextAttemptSettings);
+    boolean retryBasedOnResult =
+        shouldRetryBasedOnResult(context, previousThrowable, previousResponse);
+
+    // Short-circuit when operation has succeeded to avoid erroneously throwing
+    // CancellationException below
+    if (!retryBasedOnResult && previousThrowable == null) {
+      return false;
+    }
+
+    // may throw CancellationException if timeout is saturated
+    boolean retryBasedOnTiming = shouldRetryBasedOnTiming(context, nextAttemptSettings);
+    return retryBasedOnResult && retryBasedOnTiming;
   }
 
   boolean shouldRetryBasedOnResult(

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

@@ -31,6 +31,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -201,4 +202,36 @@ void testShouldRetryWithContext_noPreviousSettings() {
 
     assertFalse(algorithm.shouldRetry(context, previousThrowable, previousResult, null));
   }
+
+  @Test
+  void testShouldRetry_resultFalseAndThrowableNotNull_callsTimed() {
+    ResultRetryAlgorithm<Object> resultAlgorithm = mock(ResultRetryAlgorithm.class);
+    TimedRetryAlgorithm timedAlgorithm = mock(TimedRetryAlgorithm.class);
+    RetryAlgorithm<Object> algorithm = new RetryAlgorithm<>(resultAlgorithm, timedAlgorithm);
+    Throwable previousThrowable = new Throwable();
+    Object previousResult = new Object();
+    TimedAttemptSettings previousSettings = mock(TimedAttemptSettings.class);
+    when(resultAlgorithm.shouldRetry(previousThrowable, previousResult)).thenReturn(false);
+
+    boolean shouldRetry =
+        algorithm.shouldRetry(previousThrowable, previousResult, previousSettings);
+
+    assertFalse(shouldRetry);
+    verify(timedAlgorithm).shouldRetry(previousSettings);
+  }
+
+  @Test
+  void testShouldRetry_resultFalseAndThrowableNull_shortCircuits() {
+    ResultRetryAlgorithm<Object> resultAlgorithm = mock(ResultRetryAlgorithm.class);
+    TimedRetryAlgorithm timedAlgorithm = mock(TimedRetryAlgorithm.class);
+    RetryAlgorithm<Object> algorithm = new RetryAlgorithm<>(resultAlgorithm, timedAlgorithm);
+    Object previousResult = new Object();
+    TimedAttemptSettings previousSettings = mock(TimedAttemptSettings.class);
+    when(resultAlgorithm.shouldRetry(null, previousResult)).thenReturn(false);
+
+    boolean shouldRetry = algorithm.shouldRetry(null, previousResult, previousSettings);
+
+    assertFalse(shouldRetry);
+    verify(timedAlgorithm, never()).shouldRetry(previousSettings);
+  }
 }