Repository: GoogleCloudPlatform/cloud-sql-jdbc-socket-factory (java)

Issue:
Add retries around 50X errors
Port https://github.com/GoogleCloudPlatform/cloud-sql-go-connector/pull/781 to Java Connector.

We have seen the connect API occasionally give 502 errors, we should be retrying 50X errors to make the Java Connector more resilient.

Pull request:
feat: Retry API calls that return a 500 error
This updates the retry logic for API calls made by the connector: 

- Attempts to get an authentication token from the Google auth API are always retried 5 times.
- SQL Admin API requests are retried up to 5 times if the http response status code is a 5xx. Otherwise they are fatal on the first error.

This also implements the exponential backoff logic defined used in the go connector.
See: https://github.com/GoogleCloudPlatform/cloud-sql-go-connector/pull/781

Patch:
diff --git a/core/src/main/java/com/google/cloud/sql/core/ApiClientRetryingCallable.java b/core/src/main/java/com/google/cloud/sql/core/ApiClientRetryingCallable.java
new file mode 100644
index 00000000..97205943
--- /dev/null
+++ b/core/src/main/java/com/google/cloud/sql/core/ApiClientRetryingCallable.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.sql.core;
+
+import com.google.api.client.http.HttpResponseException;
+import java.util.concurrent.Callable;
+
+/**
+ * Extends RetryingCallable with logic to only retry on HTTP errors with error codes in the 5xx
+ * range.
+ *
+ * @param <T> the return value for Callable
+ */
+class ApiClientRetryingCallable<T> extends RetryingCallable<T> {
+
+  /**
+   * Construct a new RetryLogic.
+   *
+   * @param callable the callable that should be retried
+   */
+  public ApiClientRetryingCallable(Callable<T> callable) {
+    super(callable);
+  }
+
+  /**
+   * Returns false indicating that there should be another attempt if the exception is an HTTP
+   * response with an error code in the 5xx range.
+   *
+   * @param e the exception
+   * @return false if this is a http response with a 5xx status code, otherwise true.
+   */
+  @Override
+  protected boolean isFatalException(Exception e) {
+    // Only retry if the error is an HTTP response with a 5xx error code.
+    if (e instanceof HttpResponseException) {
+      HttpResponseException re = (HttpResponseException) e;
+      return re.getStatusCode() < 500;
+    }
+    // Otherwise this is a fatal exception, no more tries.
+    return true;
+  }
+}
diff --git a/core/src/main/java/com/google/cloud/sql/core/DefaultAccessTokenSupplier.java b/core/src/main/java/com/google/cloud/sql/core/DefaultAccessTokenSupplier.java
index a46cf997..d7c7260a 100644
--- a/core/src/main/java/com/google/cloud/sql/core/DefaultAccessTokenSupplier.java
+++ b/core/src/main/java/com/google/cloud/sql/core/DefaultAccessTokenSupplier.java
@@ -21,7 +21,6 @@ import com.google.auth.oauth2.GoogleCredentials;
 import com.google.cloud.sql.AuthType;
 import com.google.cloud.sql.CredentialFactory;
 import java.io.IOException;
-import java.time.Duration;
 import java.time.Instant;
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
@@ -41,8 +40,6 @@ class DefaultAccessTokenSupplier implements AccessTokenSupplier {
   private static final String SQL_LOGIN_SCOPE = "https://www.googleapis.com/auth/sqlservice.login";
 
   private final CredentialFactory credentialFactory;
-  private final int retryCount;
-  private final Duration retryDuration;
 
   static AccessTokenSupplier newInstance(AuthType authType, CredentialFactory tokenSourceFactory) {
     if (authType == AuthType.IAM) {
@@ -52,27 +49,13 @@ class DefaultAccessTokenSupplier implements AccessTokenSupplier {
     }
   }
 
-  /**
-   * Creates an instance with default retry settings.
-   *
-   * @param tokenSource the token source that produces auth tokens.
-   */
-  DefaultAccessTokenSupplier(CredentialFactory tokenSource) {
-    this(tokenSource, 3, Duration.ofSeconds(3));
-  }
-
   /**
    * Creates an instance with configurable retry settings.
    *
    * @param tokenSource the token source
-   * @param retryCount the number of attempts to refresh.
-   * @param retryDuration the duration to wait between attempts.
    */
-  DefaultAccessTokenSupplier(
-      CredentialFactory tokenSource, int retryCount, Duration retryDuration) {
+  DefaultAccessTokenSupplier(CredentialFactory tokenSource) {
     this.credentialFactory = tokenSource;
-    this.retryCount = retryCount;
-    this.retryDuration = retryDuration;
   }
 
   /**
@@ -141,9 +124,7 @@ class DefaultAccessTokenSupplier implements AccessTokenSupplier {
               }
 
               return Optional.of(downscoped.getAccessToken());
-            },
-            this.retryCount,
-            this.retryDuration);
+            });
 
     try {
       return retries.call();
diff --git a/core/src/main/java/com/google/cloud/sql/core/DefaultConnectionInfoRepository.java b/core/src/main/java/com/google/cloud/sql/core/DefaultConnectionInfoRepository.java
index ff77fc76..cb141848 100644
--- a/core/src/main/java/com/google/cloud/sql/core/DefaultConnectionInfoRepository.java
+++ b/core/src/main/java/com/google/cloud/sql/core/DefaultConnectionInfoRepository.java
@@ -112,6 +112,7 @@ class DefaultConnectionInfoRepository implements ConnectionInfoRepository {
     InstanceMetadata metadata = fetchMetadata(instanceName, authType);
     Certificate ephemeralCertificate =
         fetchEphemeralCertificate(keyPair, instanceName, token, authType);
+
     SslData sslContext =
         createSslData(keyPair, metadata, ephemeralCertificate, instanceName, authType);
 
@@ -228,10 +229,13 @@ class DefaultConnectionInfoRepository implements ConnectionInfoRepository {
   private InstanceMetadata fetchMetadata(CloudSqlInstanceName instanceName, AuthType authType) {
     try {
       ConnectSettings instanceMetadata =
-          apiClient
-              .connect()
-              .get(instanceName.getProjectId(), instanceName.getInstanceId())
-              .execute();
+          new ApiClientRetryingCallable<>(
+                  () ->
+                      apiClient
+                          .connect()
+                          .get(instanceName.getProjectId(), instanceName.getInstanceId())
+                          .execute())
+              .call();
 
       // Validate the instance will support the authenticated connection.
       if (!instanceMetadata.getRegion().equals(instanceName.getRegionId())) {
@@ -293,7 +297,7 @@ class DefaultConnectionInfoRepository implements ConnectionInfoRepository {
                 instanceName.getConnectionName()),
             ex);
       }
-    } catch (IOException ex) {
+    } catch (Exception ex) {
       throw addExceptionContext(
           ex,
           String.format(
@@ -326,12 +330,15 @@ class DefaultConnectionInfoRepository implements ConnectionInfoRepository {
     GenerateEphemeralCertResponse response;
     try {
       response =
-          apiClient
-              .connect()
-              .generateEphemeralCert(
-                  instanceName.getProjectId(), instanceName.getInstanceId(), request)
-              .execute();
-    } catch (IOException ex) {
+          new ApiClientRetryingCallable<>(
+                  () ->
+                      apiClient
+                          .connect()
+                          .generateEphemeralCert(
+                              instanceName.getProjectId(), instanceName.getInstanceId(), request)
+                          .execute())
+              .call();
+    } catch (Exception ex) {
       throw addExceptionContext(
           ex,
           String.format(
@@ -425,7 +432,7 @@ class DefaultConnectionInfoRepository implements ConnectionInfoRepository {
    *     provided to the user
    */
   private RuntimeException addExceptionContext(
-      IOException ex, String fallbackDesc, CloudSqlInstanceName instanceName) {
+      Exception ex, String fallbackDesc, CloudSqlInstanceName instanceName) {
     String reason = fallbackDesc;
     int statusCode = 0;
 
diff --git a/core/src/main/java/com/google/cloud/sql/core/RetryingCallable.java b/core/src/main/java/com/google/cloud/sql/core/RetryingCallable.java
index 6bd690e5..74505072 100644
--- a/core/src/main/java/com/google/cloud/sql/core/RetryingCallable.java
+++ b/core/src/main/java/com/google/cloud/sql/core/RetryingCallable.java
@@ -16,7 +16,6 @@
 
 package com.google.cloud.sql.core;
 
-import java.time.Duration;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ThreadLocalRandom;
 
@@ -25,59 +24,57 @@ import java.util.concurrent.ThreadLocalRandom;
  * The sleep duration is chosen randomly in the range [sleepDuration, sleepDuration * 2] to avoid
  * causing a thundering herd of requests on failure.
  *
+ * <p>exponentialBackoff calculates a duration based on the attempt i.
+ *
+ * <p>The formula is: base * multi^(attempt + 1 + random)
+ *
+ * <p>With base = 200ms and multi = 1.618, and random = [0.0, 1.0), the backoff values would fall
+ * between the following low and high ends:
+ *
+ * <p>Attempt Low (ms) High (ms)
+ *
+ * <p>0 324 524 1 524 847 2 847 1371 3 1371 2218 4 2218 3588
+ *
+ * <p>The theoretical worst case scenario would have a client wait 8.5s in total for an API request
+ * to complete (with the first four attempts failing, and the fifth succeeding).
+ *
+ * <p>This backoff strategy matches the behavior of the Cloud SQL Proxy v1.
+ *
  * @param <T> the result type of the Callable.
  */
 class RetryingCallable<T> implements Callable<T> {
+  private static final int RETRY_COUNT = 5;
 
   /** The callable that should be retried. */
   private final Callable<T> callable;
-  /** The number of times to attempt to retry. */
-  private final int retryCount;
-  /** The duration to sleep after a failed retry attempt. */
-  private final Duration sleepDuration;
 
   /**
    * Construct a new RetryLogic.
    *
    * @param callable the callable that should be retried
-   * @param retryCount the number of times to retry
-   * @param sleepDuration the duration wait after a failed attempt.
    */
-  public RetryingCallable(Callable<T> callable, int retryCount, Duration sleepDuration) {
-    if (retryCount <= 0) {
-      throw new IllegalArgumentException("retryCount must be > 0");
-    }
-    if (sleepDuration.isNegative() || sleepDuration.isZero()) {
-      throw new IllegalArgumentException("sleepDuration must be positive");
-    }
+  public RetryingCallable(Callable<T> callable) {
     if (callable == null) {
       throw new IllegalArgumentException("call must not be null");
     }
     this.callable = callable;
-    this.retryCount = retryCount;
-    this.sleepDuration = sleepDuration;
   }
 
   @Override
   public T call() throws Exception {
 
-    for (int retriesLeft = retryCount - 1; retriesLeft >= 0; retriesLeft--) {
+    for (int attempt = 0; attempt < RETRY_COUNT; attempt++) {
       // Attempt to call the Callable.
       try {
         return callable.call();
       } catch (Exception e) {
-        // Callable threw an exception.
-
-        // If this is the last iteration, then
-        // throw the exception
-        if (retriesLeft == 0) {
+        // If this is the last retry attempt, or if the exception is fatal
+        // then exit immediately.
+        if (attempt == (RETRY_COUNT - 1) || isFatalException(e)) {
           throw e;
         }
-
         // Else, sleep a random amount of time, then retry
-        long sleep =
-            ThreadLocalRandom.current()
-                .nextLong(sleepDuration.toMillis(), sleepDuration.toMillis() * 2);
+        long sleep = exponentialBackoffMs(attempt);
         try {
           Thread.sleep(sleep);
         } catch (InterruptedException ie) {
@@ -90,4 +87,15 @@ class RetryingCallable<T> implements Callable<T> {
     // as long as the preconditions in the constructor are properly met.
     throw new RuntimeException("call was never called.");
   }
+
+  protected boolean isFatalException(Exception e) {
+    return false;
+  }
+
+  private long exponentialBackoffMs(int attempt) {
+    long baseMs = 200;
+    double multi = 1.618;
+    double exp = attempt + 1.0 + ThreadLocalRandom.current().nextDouble();
+    return (long) (baseMs * Math.pow(multi, exp));
+  }
 }


Test patch:
diff --git a/core/src/test/java/com/google/cloud/sql/core/ApiClientRetryingCallableTest.java b/core/src/test/java/com/google/cloud/sql/core/ApiClientRetryingCallableTest.java
new file mode 100644
index 00000000..f9f5e8c5
--- /dev/null
+++ b/core/src/test/java/com/google/cloud/sql/core/ApiClientRetryingCallableTest.java
@@ -0,0 +1,91 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.sql.core;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.api.client.http.HttpHeaders;
+import com.google.api.client.http.HttpResponseException;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ApiClientRetryingCallableTest {
+  @Test
+  public void testApiClientRetriesOn500ErrorAndSucceeds() throws Exception {
+    AtomicInteger counter = new AtomicInteger(0);
+    ApiClientRetryingCallable<Integer> c =
+        new ApiClientRetryingCallable<>(
+            () -> {
+              int attempt = counter.incrementAndGet();
+              if (attempt < 3) {
+                throw new HttpResponseException.Builder(
+                        503, "service unavailable", new HttpHeaders())
+                    .build();
+              }
+              return attempt;
+            });
+
+    Integer v = c.call();
+    assertThat(counter.get()).isEqualTo(3);
+    assertThat(v).isEqualTo(3);
+  }
+
+  @Test
+  public void testApiClientRetriesOn500ErrorAndFailsAfter5Attempts() throws Exception {
+    AtomicInteger counter = new AtomicInteger(0);
+    ApiClientRetryingCallable<Integer> c =
+        new ApiClientRetryingCallable<>(
+            () -> {
+              counter.incrementAndGet();
+              throw new HttpResponseException.Builder(503, "service unavailable", new HttpHeaders())
+                  .build();
+            });
+
+    try {
+      c.call();
+      Assert.fail("got no exception, wants an exception to be thrown");
+    } catch (Exception e) {
+      // Expected to throw an exception
+    }
+    assertThat(counter.get()).isEqualTo(5);
+  }
+
+  @Test
+  public void testRetryStopsAfterFatalException() throws Exception {
+    final AtomicInteger counter = new AtomicInteger();
+    RetryingCallable<Integer> r =
+        new RetryingCallable<Integer>(
+            () -> {
+              counter.incrementAndGet();
+              throw new Exception("nope");
+            }) {
+          @Override
+          protected boolean isFatalException(Exception e) {
+            return true;
+          }
+        };
+
+    try {
+      r.call();
+      Assert.fail("got no exception, wants an exception to be thrown");
+    } catch (Exception e) {
+      // Expected to throw an exception
+    }
+    assertThat(counter.get()).isEqualTo(1);
+  }
+}
diff --git a/core/src/test/java/com/google/cloud/sql/core/CloudSqlCoreTestingBase.java b/core/src/test/java/com/google/cloud/sql/core/CloudSqlCoreTestingBase.java
index ffc371f9..e4f2d241 100644
--- a/core/src/test/java/com/google/cloud/sql/core/CloudSqlCoreTestingBase.java
+++ b/core/src/test/java/com/google/cloud/sql/core/CloudSqlCoreTestingBase.java
@@ -42,6 +42,7 @@ import java.security.KeyPair;
 import java.time.Duration;
 import java.util.Base64;
 import java.util.Collections;
+import java.util.concurrent.ConcurrentHashMap;
 import org.junit.Before;
 
 public class CloudSqlCoreTestingBase {
@@ -65,7 +66,7 @@ public class CloudSqlCoreTestingBase {
   public CloudSqlCoreTestingBase() {}
 
   // Creates a fake "accessNotConfigured" exception that can be used for testing.
-  static HttpTransport fakeNotConfiguredException() {
+  static MockHttpTransport fakeNotConfiguredException() {
     return fakeGoogleJsonResponseException(
         "accessNotConfigured",
         "Cloud SQL Admin API has not been used in project 12345 before or it is disabled. Enable"
@@ -77,19 +78,19 @@ public class CloudSqlCoreTestingBase {
   }
 
   // Creates a fake "notAuthorized" exception that can be used for testing.
-  static HttpTransport fakeNotAuthorizedException() {
+  static MockHttpTransport fakeNotAuthorizedException() {
     return fakeGoogleJsonResponseException(
         "notAuthorized", ERROR_MESSAGE_NOT_AUTHORIZED, HttpStatusCodes.STATUS_CODE_UNAUTHORIZED);
   }
 
   // Creates a fake "serverError" exception that can be used for testing.
-  static HttpTransport fakeBadGatewayException() {
+  static MockHttpTransport fakeBadGatewayException() {
     return fakeGoogleJsonResponseException(
         "serverError", ERROR_MESSAGE_BAD_GATEWAY, HttpStatusCodes.STATUS_CODE_BAD_GATEWAY);
   }
 
   // Builds a fake GoogleJsonResponseException for testing API error handling.
-  private static HttpTransport fakeGoogleJsonResponseException(
+  private static MockHttpTransport fakeGoogleJsonResponseException(
       String reason, String message, int statusCode) {
     ErrorInfo errorInfo = new ErrorInfo();
     errorInfo.setReason(reason);
@@ -97,7 +98,7 @@ public class CloudSqlCoreTestingBase {
     return fakeGoogleJsonResponseExceptionTransport(errorInfo, message, statusCode);
   }
 
-  private static HttpTransport fakeGoogleJsonResponseExceptionTransport(
+  private static MockHttpTransport fakeGoogleJsonResponseExceptionTransport(
       ErrorInfo errorInfo, String message, int statusCode) {
     final JsonFactory jsonFactory = new GsonFactory();
     return new MockHttpTransport() {
@@ -117,7 +118,7 @@ public class CloudSqlCoreTestingBase {
                 new MockLowLevelHttpResponse()
                     .setContent(errorResponse.toPrettyString())
                     .setContentType(Json.MEDIA_TYPE)
-                    .setStatusCode(HttpStatusCodes.STATUS_CODE_FORBIDDEN));
+                    .setStatusCode(statusCode));
       }
     };
   }
@@ -131,19 +132,20 @@ public class CloudSqlCoreTestingBase {
     clientKeyPair = Futures.immediateFuture(TestKeys.getClientKeyPair());
   }
 
-  HttpTransport fakeSuccessHttpTransport(Duration certDuration) {
+  MockHttpTransport fakeSuccessHttpTransport(Duration certDuration) {
     return fakeSuccessHttpTransport(TestKeys.getServerCertPem(), certDuration, null);
   }
 
-  HttpTransport fakeSuccessHttpTransport(Duration certDuration, String baseUrl) {
+  MockHttpTransport fakeSuccessHttpTransport(Duration certDuration, String baseUrl) {
     return fakeSuccessHttpTransport(TestKeys.getServerCertPem(), certDuration, baseUrl);
   }
 
-  HttpTransport fakeSuccessHttpTransport(String serverCert, Duration certDuration) {
+  MockHttpTransport fakeSuccessHttpTransport(String serverCert, Duration certDuration) {
     return fakeSuccessHttpTransport(serverCert, certDuration, null);
   }
 
-  HttpTransport fakeSuccessHttpTransport(String serverCert, Duration certDuration, String baseUrl) {
+  MockHttpTransport fakeSuccessHttpTransport(
+      String serverCert, Duration certDuration, String baseUrl) {
     final JsonFactory jsonFactory = new GsonFactory();
     return new MockHttpTransport() {
       @Override
@@ -187,4 +189,34 @@ public class CloudSqlCoreTestingBase {
       }
     };
   }
+
+  HttpTransport fakeIntermittentErrorHttpTransport(
+      MockHttpTransport successTransport, MockHttpTransport errorTransport) {
+    // Odd request counts get errors, Even request count gets success.
+    ConcurrentHashMap<String, Integer> counterByUrl = new ConcurrentHashMap<>();
+
+    return new MockHttpTransport() {
+      @Override
+      public LowLevelHttpRequest buildRequest(String method, String url) {
+        return new MockLowLevelHttpRequest() {
+          @Override
+          public LowLevelHttpResponse execute() throws IOException {
+            int count =
+                counterByUrl.compute(
+                    url,
+                    (url, val) -> {
+                      if (val == null) {
+                        return 1;
+                      }
+                      return val + 1;
+                    });
+            if (count % 2 == 0) {
+              return successTransport.buildRequest(method, url).execute();
+            }
+            return errorTransport.buildRequest(method, url).execute();
+          }
+        };
+      }
+    };
+  }
 }
diff --git a/core/src/test/java/com/google/cloud/sql/core/ConnectorTest.java b/core/src/test/java/com/google/cloud/sql/core/ConnectorTest.java
index 93caf480..3e8e6028 100644
--- a/core/src/test/java/com/google/cloud/sql/core/ConnectorTest.java
+++ b/core/src/test/java/com/google/cloud/sql/core/ConnectorTest.java
@@ -337,15 +337,51 @@ public class ConnectorTest extends CloudSqlCoreTestingBase {
             TEST_MAX_REFRESH_MS,
             DEFAULT_SERVER_PROXY_PORT);
 
-    TerminalException ex =
-        assertThrows(TerminalException.class, () -> c.connect(config, TEST_MAX_REFRESH_MS));
+    // If the gateway is down, then this is a temporary error, not a fatal error.
+    RuntimeException ex =
+        assertThrows(RuntimeException.class, () -> c.connect(config, TEST_MAX_REFRESH_MS));
 
-    assertThat(ex)
-        .hasMessageThat()
-        .contains(
-            String.format(
-                "[%s] The Google Cloud SQL Admin API failed for the project",
-                "myProject:myRegion:NotMyInstance"));
+    // The Connector.connect() method will timeout, and will include details about the instance
+    // data in the test.
+    assertThat(ex).hasMessageThat().contains("Unable to get valid instance data within");
+
+    assertThat(ex).hasMessageThat().contains("502");
+
+    assertThat(ex).hasMessageThat().contains("myProject:myRegion:NotMyInstance");
+  }
+
+  @Test
+  public void create_successfulPublicConnection_withIntermittentBadGatewayErrors()
+      throws IOException, InterruptedException {
+    ConnectionInfoRepositoryFactory factory =
+        new StubConnectionInfoRepositoryFactory(
+            fakeIntermittentErrorHttpTransport(
+                fakeSuccessHttpTransport(Duration.ofSeconds(0)), fakeBadGatewayException()));
+
+    FakeSslServer sslServer = new FakeSslServer();
+
+    ConnectionConfig config =
+        new ConnectionConfig.Builder()
+            .withCloudSqlInstance("myProject:myRegion:myInstance")
+            .withIpTypes("PRIMARY")
+            .build();
+
+    int port = sslServer.start(PUBLIC_IP);
+
+    Connector c =
+        new Connector(
+            config.getConnectorConfig(),
+            factory,
+            stubCredentialFactoryProvider.getInstanceCredentialFactory(config.getConnectorConfig()),
+            defaultExecutor,
+            clientKeyPair,
+            10,
+            TEST_MAX_REFRESH_MS,
+            port);
+
+    Socket socket = c.connect(config, TEST_MAX_REFRESH_MS);
+
+    assertThat(readLine(socket)).isEqualTo(SERVER_MESSAGE);
   }
 
   @Test
diff --git a/core/src/test/java/com/google/cloud/sql/core/DefaultAccessTokenSupplierTest.java b/core/src/test/java/com/google/cloud/sql/core/DefaultAccessTokenSupplierTest.java
index c1216b31..52ae166b 100644
--- a/core/src/test/java/com/google/cloud/sql/core/DefaultAccessTokenSupplierTest.java
+++ b/core/src/test/java/com/google/cloud/sql/core/DefaultAccessTokenSupplierTest.java
@@ -37,7 +37,6 @@ import com.google.auth.oauth2.GoogleCredentials;
 import com.google.auth.oauth2.OAuth2Credentials;
 import com.google.cloud.sql.CredentialFactory;
 import java.io.IOException;
-import java.time.Duration;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.Collections;
@@ -75,8 +74,7 @@ public class DefaultAccessTokenSupplierTest {
 
   @Test
   public void testEmptyTokenOnEmptyCredentials() throws IOException {
-    DefaultAccessTokenSupplier supplier =
-        new DefaultAccessTokenSupplier(null, 1, Duration.ofMillis(10));
+    DefaultAccessTokenSupplier supplier = new DefaultAccessTokenSupplier(null);
     assertThat(supplier.get()).isEqualTo(Optional.empty());
   }
 
@@ -100,8 +98,7 @@ public class DefaultAccessTokenSupplierTest {
         };
 
     DefaultAccessTokenSupplier supplier =
-        new DefaultAccessTokenSupplier(
-            new GoogleCredentialsFactory(googleCredentials), 1, Duration.ofMillis(10));
+        new DefaultAccessTokenSupplier(new GoogleCredentialsFactory(googleCredentials));
     Optional<AccessToken> token = supplier.get();
 
     assertThat(token.isPresent()).isTrue();
@@ -127,8 +124,7 @@ public class DefaultAccessTokenSupplierTest {
           }
         };
 
-    DefaultAccessTokenSupplier supplier =
-        new DefaultAccessTokenSupplier(badFactory, 1, Duration.ofMillis(10));
+    DefaultAccessTokenSupplier supplier = new DefaultAccessTokenSupplier(badFactory);
     RuntimeException ex = assertThrows(RuntimeException.class, supplier::get);
     assertThat(ex).hasMessageThat().contains("Unsupported credentials of type");
   }
@@ -153,11 +149,10 @@ public class DefaultAccessTokenSupplierTest {
         };
 
     DefaultAccessTokenSupplier supplier =
-        new DefaultAccessTokenSupplier(
-            new GoogleCredentialsFactory(expiredGoogleCredentials), 1, Duration.ofMillis(10));
+        new DefaultAccessTokenSupplier(new GoogleCredentialsFactory(expiredGoogleCredentials));
     IllegalStateException ex = assertThrows(IllegalStateException.class, supplier::get);
     assertThat(ex).hasMessageThat().contains("Error refreshing credentials");
-    assertThat(refreshCounter.get()).isEqualTo(1);
+    assertThat(refreshCounter.get()).isEqualTo(5);
   }
 
   @Test
@@ -180,11 +175,10 @@ public class DefaultAccessTokenSupplierTest {
         };
 
     DefaultAccessTokenSupplier supplier =
-        new DefaultAccessTokenSupplier(
-            new GoogleCredentialsFactory(refreshGetsExpiredToken), 1, Duration.ofMillis(10));
+        new DefaultAccessTokenSupplier(new GoogleCredentialsFactory(refreshGetsExpiredToken));
     IllegalStateException ex = assertThrows(IllegalStateException.class, supplier::get);
     assertThat(ex).hasMessageThat().contains("expiration time is in the past");
-    assertThat(refreshCounter.get()).isEqualTo(1);
+    assertThat(refreshCounter.get()).isEqualTo(5);
   }
 
   @Test
@@ -206,8 +200,7 @@ public class DefaultAccessTokenSupplierTest {
         };
 
     DefaultAccessTokenSupplier supplier =
-        new DefaultAccessTokenSupplier(
-            new GoogleCredentialsFactory(refreshableCredentials), 1, Duration.ofMillis(10));
+        new DefaultAccessTokenSupplier(new GoogleCredentialsFactory(refreshableCredentials));
     Optional<AccessToken> token = supplier.get();
 
     assertThat(token.isPresent()).isTrue();
@@ -239,8 +232,7 @@ public class DefaultAccessTokenSupplierTest {
         };
 
     DefaultAccessTokenSupplier supplier =
-        new DefaultAccessTokenSupplier(
-            new GoogleCredentialsFactory(refreshableCredentials), 3, Duration.ofMillis(10));
+        new DefaultAccessTokenSupplier(new GoogleCredentialsFactory(refreshableCredentials));
     Optional<AccessToken> token = supplier.get();
 
     assertThat(token.isPresent()).isTrue();
@@ -271,8 +263,7 @@ public class DefaultAccessTokenSupplierTest {
   public void throwsErrorForWrongCredentialType() {
     OAuth2Credentials creds = OAuth2Credentials.create(new AccessToken("abc", null));
     DefaultAccessTokenSupplier supplier =
-        new DefaultAccessTokenSupplier(
-            new Oauth2BadCredentialFactory(creds), 1, Duration.ofMillis(10));
+        new DefaultAccessTokenSupplier(new Oauth2BadCredentialFactory(creds));
     RuntimeException ex = assertThrows(RuntimeException.class, supplier::get);
 
     assertThat(ex)
@@ -291,8 +282,7 @@ public class DefaultAccessTokenSupplierTest {
           }
         };
     DefaultAccessTokenSupplier supplier =
-        new DefaultAccessTokenSupplier(
-            new GoogleCredentialsFactory(creds), 1, Duration.ofMillis(10));
+        new DefaultAccessTokenSupplier(new GoogleCredentialsFactory(creds));
     RuntimeException ex = assertThrows(RuntimeException.class, supplier::get);
 
     assertThat(ex).hasMessageThat().contains("Access Token has length of zero");
@@ -317,8 +307,7 @@ public class DefaultAccessTokenSupplierTest {
         };
 
     DefaultAccessTokenSupplier supplier =
-        new DefaultAccessTokenSupplier(
-            new GoogleCredentialsFactory(refreshableCredentials), 1, Duration.ofMillis(10));
+        new DefaultAccessTokenSupplier(new GoogleCredentialsFactory(refreshableCredentials));
     RuntimeException ex = assertThrows(RuntimeException.class, supplier::get);
 
     assertThat(ex).hasMessageThat().contains("Access Token expiration time is in the past");
@@ -368,8 +357,7 @@ public class DefaultAccessTokenSupplierTest {
     credential.setExpirationTimeMilliseconds(future.toEpochMilli());
 
     DefaultAccessTokenSupplier supplier =
-        new DefaultAccessTokenSupplier(
-            new Oauth2CredentialFactory(credential), 1, Duration.ofMillis(10));
+        new DefaultAccessTokenSupplier(new Oauth2CredentialFactory(credential));
     Optional<AccessToken> token = supplier.get();
 
     assertThat(token.isPresent()).isTrue();
@@ -438,8 +426,7 @@ public class DefaultAccessTokenSupplierTest {
     credential.setExpirationTimeMilliseconds(past.toEpochMilli());
 
     DefaultAccessTokenSupplier supplier =
-        new DefaultAccessTokenSupplier(
-            new Oauth2CredentialFactory(credential), 1, Duration.ofMillis(10));
+        new DefaultAccessTokenSupplier(new Oauth2CredentialFactory(credential));
     Optional<AccessToken> token = supplier.get();
 
     assertThat(token.isPresent()).isTrue();
diff --git a/core/src/test/java/com/google/cloud/sql/core/RetryingCallableTest.java b/core/src/test/java/com/google/cloud/sql/core/RetryingCallableTest.java
index c11f056f..b3d34c8c 100644
--- a/core/src/test/java/com/google/cloud/sql/core/RetryingCallableTest.java
+++ b/core/src/test/java/com/google/cloud/sql/core/RetryingCallableTest.java
@@ -18,7 +18,6 @@ package com.google.cloud.sql.core;
 
 import static com.google.common.truth.Truth.assertThat;
 
-import java.time.Duration;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Assert;
 import org.junit.Test;
@@ -27,30 +26,12 @@ public class RetryingCallableTest {
   @Test
   public void testConstructorIllegalArguments() throws Exception {
     // Callable must not be null
-    Assert.assertThrows(
-        IllegalArgumentException.class,
-        () -> new RetryingCallable<>(null, 1, Duration.ofMillis(1)));
-
-    // Must have a positive retryCount
-    Assert.assertThrows(
-        IllegalArgumentException.class,
-        () -> new RetryingCallable<>(() -> null, 0, Duration.ofMillis(1)));
-    Assert.assertThrows(
-        IllegalArgumentException.class,
-        () -> new RetryingCallable<>(() -> null, -1, Duration.ofMillis(1)));
-
-    // Must have a positive duration
-    Assert.assertThrows(
-        IllegalArgumentException.class,
-        () -> new RetryingCallable<>(() -> null, 2, Duration.ofMillis(-5)));
-    Assert.assertThrows(
-        IllegalArgumentException.class,
-        () -> new RetryingCallable<>(() -> null, 2, Duration.ofMillis(0)));
+    Assert.assertThrows(IllegalArgumentException.class, () -> new RetryingCallable<>(null));
   }
 
   @Test
   public void testNoRetryRequired() throws Exception {
-    RetryingCallable<Integer> r = new RetryingCallable<>(() -> 1, 5, Duration.ofMillis(100));
+    RetryingCallable<Integer> r = new RetryingCallable<>(() -> 1);
     int v = r.call();
     assertThat(v).isEqualTo(1);
   }
@@ -63,9 +44,7 @@ public class RetryingCallableTest {
             () -> {
               counter.incrementAndGet();
               throw new Exception("nope");
-            },
-            3,
-            Duration.ofMillis(100));
+            });
 
     try {
       r.call();
@@ -74,7 +53,7 @@ public class RetryingCallableTest {
       // Expected to throw an exception
     }
 
-    assertThat(counter.get()).isEqualTo(3);
+    assertThat(counter.get()).isEqualTo(5);
   }
 
   @Test
@@ -88,12 +67,37 @@ public class RetryingCallableTest {
                 throw new Exception("nope");
               }
               return i;
-            },
-            5,
-            Duration.ofMillis(100));
+            });
 
     int v = r.call();
     assertThat(counter.get()).isEqualTo(3);
     assertThat(v).isEqualTo(3);
   }
+
+  @Test
+  public void testRetryStopsAfterFatalException() throws Exception {
+    final AtomicInteger counter = new AtomicInteger();
+    RetryingCallable<Integer> r =
+        new RetryingCallable<Integer>(
+            () -> {
+              int i = counter.incrementAndGet();
+              if (i < 3) {
+                throw new Exception("nope");
+              }
+              return i;
+            }) {
+          @Override
+          protected boolean isFatalException(Exception e) {
+            return true;
+          }
+        };
+
+    try {
+      r.call();
+      Assert.fail("got no exception, wants an exception to be thrown");
+    } catch (Exception e) {
+      // Expected to throw an exception
+    }
+    assertThat(counter.get()).isEqualTo(1);
+  }
 }