Repository: GoogleCloudPlatform/cloud-sql-jdbc-socket-factory
Pull request #1981 (2024-05-15 15:41:56)

chore: Add synchronous operation to DefaultConnectionInfoRepository (part of #992)
Adds a new method ConnectionInfoRepository.getConnectorInfoSync() which blocks until it can return a valid 
ConnectorInfo. Adds the implementation of this new method to DefaultConnectionInfoRepository.  This is used 
by the LazyRefresher to fetch connection info on the active thread.  

Part of Lazy Refresh #992.


Patch:
diff --git a/core/src/main/java/com/google/cloud/sql/core/ConnectionInfoRepository.java b/core/src/main/java/com/google/cloud/sql/core/ConnectionInfoRepository.java
index 4882b543..8039cd88 100644
--- a/core/src/main/java/com/google/cloud/sql/core/ConnectionInfoRepository.java
+++ b/core/src/main/java/com/google/cloud/sql/core/ConnectionInfoRepository.java
@@ -5,7 +5,7 @@
  * 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
+ * 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,
@@ -30,4 +30,10 @@ interface ConnectionInfoRepository {
       AuthType authType,
       ListeningScheduledExecutorService executor,
       ListenableFuture<KeyPair> keyPair);
+
+  ConnectionInfo getConnectionInfoSync(
+      CloudSqlInstanceName instanceName,
+      AccessTokenSupplier accessTokenSupplier,
+      AuthType authType,
+      KeyPair keyPair);
 }
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 159eadbd..ed6719a4 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
@@ -96,6 +96,29 @@ class DefaultConnectionInfoRepository implements ConnectionInfoRepository {
         + "-----END RSA PUBLIC KEY-----\n";
   }
 
+  /** Internal Use Only: Gets the instance data for the CloudSqlInstance from the API. */
+  @Override
+  public ConnectionInfo getConnectionInfoSync(
+      CloudSqlInstanceName instanceName,
+      AccessTokenSupplier accessTokenSupplier,
+      AuthType authType,
+      KeyPair keyPair) {
+    Optional<AccessToken> token = null;
+    try {
+      token = accessTokenSupplier.get();
+    } catch (IOException e) {
+      throw new RuntimeException("Unable to create IAM Auth access token", e);
+    }
+    InstanceMetadata metadata = fetchMetadata(instanceName, authType);
+    Certificate ephemeralCertificate =
+        fetchEphemeralCertificate(keyPair, instanceName, token, authType);
+    SslData sslContext =
+        createSslData(keyPair, metadata, ephemeralCertificate, instanceName, authType);
+
+    return createConnectionInfo(
+        instanceName, authType, token, metadata, ephemeralCertificate, sslContext);
+  }
+
   /** Internal Use Only: Gets the instance data for the CloudSqlInstance from the API. */
   @Override
   public ListenableFuture<ConnectionInfo> getConnectionInfo(
@@ -137,32 +160,14 @@ class DefaultConnectionInfoRepository implements ConnectionInfoRepository {
     ListenableFuture<ConnectionInfo> done =
         Futures.whenAllComplete(metadataFuture, ephemeralCertificateFuture, sslContextFuture)
             .call(
-                () -> {
-
-                  // Get expiration value for new cert
-                  Certificate ephemeralCertificate = Futures.getDone(ephemeralCertificateFuture);
-                  X509Certificate x509Certificate = (X509Certificate) ephemeralCertificate;
-                  Instant expiration = x509Certificate.getNotAfter().toInstant();
-
-                  if (authType == AuthType.IAM) {
-                    expiration =
-                        DefaultAccessTokenSupplier.getTokenExpirationTime(Futures.getDone(token))
-                            .filter(
-                                tokenExpiration ->
-                                    x509Certificate
-                                        .getNotAfter()
-                                        .toInstant()
-                                        .isAfter(tokenExpiration))
-                            .orElse(x509Certificate.getNotAfter().toInstant());
-                  }
-
-                  logger.debug(String.format("[%s] INSTANCE DATA DONE", instanceName));
-
-                  return new ConnectionInfo(
-                      Futures.getDone(metadataFuture),
-                      Futures.getDone(sslContextFuture),
-                      expiration);
-                },
+                () ->
+                    createConnectionInfo(
+                        instanceName,
+                        authType,
+                        Futures.getDone(token),
+                        Futures.getDone(metadataFuture),
+                        Futures.getDone(ephemeralCertificateFuture),
+                        Futures.getDone(sslContextFuture)),
                 executor);
 
     done.addListener(
@@ -170,6 +175,31 @@ class DefaultConnectionInfoRepository implements ConnectionInfoRepository {
     return done;
   }
 
+  private static ConnectionInfo createConnectionInfo(
+      CloudSqlInstanceName instanceName,
+      AuthType authType,
+      Optional<AccessToken> token,
+      InstanceMetadata metadata,
+      Certificate ephemeralCertificate,
+      SslData sslContext) {
+    // Get expiration value for new cert
+    X509Certificate x509Certificate = (X509Certificate) ephemeralCertificate;
+    Instant expiration = x509Certificate.getNotAfter().toInstant();
+
+    if (authType == AuthType.IAM) {
+      expiration =
+          DefaultAccessTokenSupplier.getTokenExpirationTime(token)
+              .filter(
+                  tokenExpiration ->
+                      x509Certificate.getNotAfter().toInstant().isAfter(tokenExpiration))
+              .orElse(x509Certificate.getNotAfter().toInstant());
+    }
+
+    logger.debug(String.format("[%s] INSTANCE DATA DONE", instanceName));
+
+    return new ConnectionInfo(metadata, sslContext, expiration);
+  }
+
   String getApplicationName() {
     return apiClient.getApplicationName();
   }


Test patch:
diff --git a/core/src/test/java/com/google/cloud/sql/core/DefaultConnectionInfoCacheTest.java b/core/src/test/java/com/google/cloud/sql/core/DefaultConnectionInfoCacheTest.java
index da9df8ff..4f879283 100644
--- a/core/src/test/java/com/google/cloud/sql/core/DefaultConnectionInfoCacheTest.java
+++ b/core/src/test/java/com/google/cloud/sql/core/DefaultConnectionInfoCacheTest.java
@@ -90,15 +90,17 @@ public class DefaultConnectionInfoCacheTest {
   public void testInstanceFailsOnConnectionError() {
 
     ConnectionInfoRepository connectionInfoRepository =
-        (CloudSqlInstanceName instanceName,
-            AccessTokenSupplier accessTokenSupplier,
-            AuthType authType,
-            ListeningScheduledExecutorService exec,
-            ListenableFuture<KeyPair> keyPair) ->
-            exec.submit(
-                () -> {
-                  throw new RuntimeException("always fails");
-                });
+        new StubConnectionInfoRepository() {
+          @Override
+          public ListenableFuture<ConnectionInfo> getConnectionInfo(
+              CloudSqlInstanceName instanceName,
+              AccessTokenSupplier accessTokenSupplier,
+              AuthType authType,
+              ListeningScheduledExecutorService executor,
+              ListenableFuture<KeyPair> keyPair) {
+            throw new RuntimeException("always fails");
+          }
+        };
 
     // initialize connectionInfoCache after mocks are set up
     DefaultConnectionInfoCache connectionInfoCache =
@@ -121,14 +123,18 @@ public class DefaultConnectionInfoCacheTest {
   public void testInstanceFailsOnTooLongToRetrieve() {
     PauseCondition cond = new PauseCondition();
     ConnectionInfoRepository connectionInfoRepository =
-        (CloudSqlInstanceName instanceName,
-            AccessTokenSupplier accessTokenSupplier,
-            AuthType authType,
-            ListeningScheduledExecutorService exec,
-            ListenableFuture<KeyPair> keyPair) -> {
-          // This is never allowed to proceed
-          cond.pause();
-          throw new RuntimeException("fake read timeout");
+        new StubConnectionInfoRepository() {
+          @Override
+          public ListenableFuture<ConnectionInfo> getConnectionInfo(
+              CloudSqlInstanceName instanceName,
+              AccessTokenSupplier accessTokenSupplier,
+              AuthType authType,
+              ListeningScheduledExecutorService executor,
+              ListenableFuture<KeyPair> keyPair) {
+            // This is never allowed to proceed
+            cond.pause();
+            throw new RuntimeException("fake read timeout");
+          }
         };
 
     // initialize connectionInfoCache after mocks are set up
@@ -151,20 +157,30 @@ public class DefaultConnectionInfoCacheTest {
     ConnectionInfo connectionInfo = newFutureConnectionInfo();
     AtomicInteger refreshCount = new AtomicInteger();
     final PauseCondition cond = new PauseCondition();
+    ConnectionInfoRepository connectionInfoRepository =
+        new StubConnectionInfoRepository() {
+          @Override
+          public ListenableFuture<ConnectionInfo> getConnectionInfo(
+              CloudSqlInstanceName instanceName,
+              AccessTokenSupplier accessTokenSupplier,
+              AuthType authType,
+              ListeningScheduledExecutorService executor,
+              ListenableFuture<KeyPair> keyPair) {
+            int c = refreshCount.get();
+            // Allow the first execution to complete immediately.
+            // The second execution should pause until signaled.
+            if (c == 1) {
+              cond.pause();
+            }
+            refreshCount.incrementAndGet();
+            return Futures.immediateFuture(connectionInfo);
+          }
+        };
 
     DefaultConnectionInfoCache connectionInfoCache =
         new DefaultConnectionInfoCache(
             new ConnectionConfig.Builder().withCloudSqlInstance("project:region:instance").build(),
-            (instanceName, accessTokenSupplier, authType, executor, keyPair) -> {
-              int c = refreshCount.get();
-              // Allow the first execution to complete immediately.
-              // The second execution should pause until signaled.
-              if (c == 1) {
-                cond.pause();
-              }
-              refreshCount.incrementAndGet();
-              return Futures.immediateFuture(connectionInfo);
-            },
+            connectionInfoRepository,
             stubCredentialFactory,
             executorService,
             keyPairFuture,
@@ -197,17 +213,27 @@ public class DefaultConnectionInfoCacheTest {
 
     AtomicInteger refreshCount = new AtomicInteger();
 
+    ConnectionInfoRepository connectionInfoRepository =
+        new StubConnectionInfoRepository() {
+          @Override
+          public ListenableFuture<ConnectionInfo> getConnectionInfo(
+              CloudSqlInstanceName instanceName,
+              AccessTokenSupplier accessTokenSupplier,
+              AuthType authType,
+              ListeningScheduledExecutorService executor,
+              ListenableFuture<KeyPair> keyPair) {
+            int c = refreshCount.get();
+            refreshCount.incrementAndGet();
+            if (c == 0) {
+              throw new RuntimeException("bad request 0");
+            }
+            return Futures.immediateFuture(connectionInfo);
+          }
+        };
     DefaultConnectionInfoCache connectionInfoCache =
         new DefaultConnectionInfoCache(
             new ConnectionConfig.Builder().withCloudSqlInstance("project:region:instance").build(),
-            (instanceName, accessTokenSupplier, authType, executor, keyPair) -> {
-              int c = refreshCount.get();
-              refreshCount.incrementAndGet();
-              if (c == 0) {
-                throw new RuntimeException("bad request 0");
-              }
-              return Futures.immediateFuture(connectionInfo);
-            },
+            connectionInfoRepository,
             stubCredentialFactory,
             executorService,
             keyPairFuture,
@@ -249,27 +275,37 @@ public class DefaultConnectionInfoCacheTest {
 
     AtomicInteger refreshCount = new AtomicInteger();
     final PauseCondition refresh1 = new PauseCondition();
-
+    ConnectionInfoRepository connectionInfoRepository =
+        new StubConnectionInfoRepository() {
+          @Override
+          public ListenableFuture<ConnectionInfo> getConnectionInfo(
+              CloudSqlInstanceName instanceName,
+              AccessTokenSupplier accessTokenSupplier,
+              AuthType authType,
+              ListeningScheduledExecutorService executor,
+              ListenableFuture<KeyPair> keyPair) {
+
+            int c = refreshCount.get();
+            ConnectionInfo refreshResult = info;
+            switch (c) {
+              case 0:
+                // refresh 0 should return initial immediately
+                refreshResult = initial;
+                break;
+              case 1:
+                // refresh 1 should pause
+                refresh1.pause();
+                break;
+            }
+            // refresh 2 and on should return data immediately
+            refreshCount.incrementAndGet();
+            return Futures.immediateFuture(refreshResult);
+          }
+        };
     DefaultConnectionInfoCache connectionInfoCache =
         new DefaultConnectionInfoCache(
             new ConnectionConfig.Builder().withCloudSqlInstance("project:region:instance").build(),
-            (instanceName, accessTokenSupplier, authType, executor, keyPair) -> {
-              int c = refreshCount.get();
-              ConnectionInfo refreshResult = info;
-              switch (c) {
-                case 0:
-                  // refresh 0 should return initial immediately
-                  refreshResult = initial;
-                  break;
-                case 1:
-                  // refresh 1 should pause
-                  refresh1.pause();
-                  break;
-              }
-              // refresh 2 and on should return data immediately
-              refreshCount.incrementAndGet();
-              return Futures.immediateFuture(refreshResult);
-            },
+            connectionInfoRepository,
             stubCredentialFactory,
             executorService,
             keyPairFuture,
@@ -311,24 +347,34 @@ public class DefaultConnectionInfoCacheTest {
     final PauseCondition refresh0 = new PauseCondition();
     final PauseCondition refresh1 = new PauseCondition();
 
+    ConnectionInfoRepository connectionInfoRepository =
+        new StubConnectionInfoRepository() {
+          @Override
+          public ListenableFuture<ConnectionInfo> getConnectionInfo(
+              CloudSqlInstanceName instanceName,
+              AccessTokenSupplier accessTokenSupplier,
+              AuthType authType,
+              ListeningScheduledExecutorService executor,
+              ListenableFuture<KeyPair> keyPair) {
+            int c = refreshCount.get();
+            ConnectionInfo refreshResult = info;
+            switch (c) {
+              case 0:
+                refresh0.pause();
+                refreshResult = expiresInOneMinute;
+                break;
+              case 1:
+                refresh1.pause();
+                break;
+            }
+            refreshCount.incrementAndGet();
+            return Futures.immediateFuture(refreshResult);
+          }
+        };
     DefaultConnectionInfoCache connectionInfoCache =
         new DefaultConnectionInfoCache(
             new ConnectionConfig.Builder().withCloudSqlInstance("project:region:instance").build(),
-            (instanceName, accessTokenSupplier, authType, executor, keyPair) -> {
-              int c = refreshCount.get();
-              ConnectionInfo refreshResult = info;
-              switch (c) {
-                case 0:
-                  refresh0.pause();
-                  refreshResult = expiresInOneMinute;
-                  break;
-                case 1:
-                  refresh1.pause();
-                  break;
-              }
-              refreshCount.incrementAndGet();
-              return Futures.immediateFuture(refreshResult);
-            },
+            connectionInfoRepository,
             stubCredentialFactory,
             executorService,
             keyPairFuture,
@@ -374,24 +420,34 @@ public class DefaultConnectionInfoCacheTest {
 
     AtomicInteger refreshCount = new AtomicInteger();
     final PauseCondition refresh1 = new PauseCondition();
+    ConnectionInfoRepository connectionInfoRepository =
+        new StubConnectionInfoRepository() {
+          @Override
+          public ListenableFuture<ConnectionInfo> getConnectionInfo(
+              CloudSqlInstanceName instanceName,
+              AccessTokenSupplier accessTokenSupplier,
+              AuthType authType,
+              ListeningScheduledExecutorService executor,
+              ListenableFuture<KeyPair> keyPair) {
+            int c = refreshCount.get();
+            switch (c) {
+              case 0:
+                refreshCount.incrementAndGet();
+                return Futures.immediateFuture(initialData);
+              case 1:
+                refresh1.pause();
+                refreshCount.incrementAndGet();
+                return Futures.immediateFuture(info);
+              default:
+                return Futures.immediateFuture(info);
+            }
+          }
+        };
 
     DefaultConnectionInfoCache connectionInfoCache =
         new DefaultConnectionInfoCache(
             new ConnectionConfig.Builder().withCloudSqlInstance("project:region:instance").build(),
-            (instanceName, accessTokenSupplier, authType, executor, keyPair) -> {
-              int c = refreshCount.get();
-              switch (c) {
-                case 0:
-                  refreshCount.incrementAndGet();
-                  return Futures.immediateFuture(initialData);
-                case 1:
-                  refresh1.pause();
-                  refreshCount.incrementAndGet();
-                  return Futures.immediateFuture(info);
-                default:
-                  return Futures.immediateFuture(info);
-              }
-            },
+            connectionInfoRepository,
             stubCredentialFactory,
             executorService,
             keyPairFuture,
@@ -435,30 +491,40 @@ public class DefaultConnectionInfoCacheTest {
     final PauseCondition badRequest1 = new PauseCondition();
     final PauseCondition badRequest2 = new PauseCondition();
     final PauseCondition goodRequest = new PauseCondition();
+    ConnectionInfoRepository connectionInfoRepository =
+        new StubConnectionInfoRepository() {
+          @Override
+          public ListenableFuture<ConnectionInfo> getConnectionInfo(
+              CloudSqlInstanceName instanceName,
+              AccessTokenSupplier accessTokenSupplier,
+              AuthType authType,
+              ListeningScheduledExecutorService executor,
+              ListenableFuture<KeyPair> keyPair) {
+            int c = refreshCount.get();
+            switch (c) {
+              case 0:
+                refreshCount.incrementAndGet();
+                return Futures.immediateFuture(aboutToExpire);
+              case 1:
+                badRequest1.pause();
+                refreshCount.incrementAndGet();
+                throw new RuntimeException("bad request 1");
+              case 2:
+                badRequest2.pause();
+                refreshCount.incrementAndGet();
+                throw new RuntimeException("bad request 2");
+              default:
+                goodRequest.pause();
+                refreshCount.incrementAndGet();
+                return Futures.immediateFuture(info);
+            }
+          }
+        };
 
     DefaultConnectionInfoCache connectionInfoCache =
         new DefaultConnectionInfoCache(
             new ConnectionConfig.Builder().withCloudSqlInstance("project:region:instance").build(),
-            (instanceName, accessTokenSupplier, authType, executor, keyPair) -> {
-              int c = refreshCount.get();
-              switch (c) {
-                case 0:
-                  refreshCount.incrementAndGet();
-                  return Futures.immediateFuture(aboutToExpire);
-                case 1:
-                  badRequest1.pause();
-                  refreshCount.incrementAndGet();
-                  throw new RuntimeException("bad request 1");
-                case 2:
-                  badRequest2.pause();
-                  refreshCount.incrementAndGet();
-                  throw new RuntimeException("bad request 2");
-                default:
-                  goodRequest.pause();
-                  refreshCount.incrementAndGet();
-                  return Futures.immediateFuture(info);
-              }
-            },
+            connectionInfoRepository,
             stubCredentialFactory,
             executorService,
             keyPairFuture,
@@ -517,9 +583,17 @@ public class DefaultConnectionInfoCacheTest {
     AtomicInteger refreshCount = new AtomicInteger();
 
     ConnectionInfoRepository connectionInfoRepository =
-        (instanceName, accessTokenSupplier, authType, executor, keyPair) -> {
-          refreshCount.incrementAndGet();
-          return Futures.immediateFuture(info);
+        new StubConnectionInfoRepository() {
+          @Override
+          public ListenableFuture<ConnectionInfo> getConnectionInfo(
+              CloudSqlInstanceName instanceName,
+              AccessTokenSupplier accessTokenSupplier,
+              AuthType authType,
+              ListeningScheduledExecutorService executor,
+              ListenableFuture<KeyPair> keyPair) {
+            refreshCount.incrementAndGet();
+            return Futures.immediateFuture(info);
+          }
         };
 
     Map<List<IpType>, String> values = new LinkedHashMap<>();
@@ -563,9 +637,17 @@ public class DefaultConnectionInfoCacheTest {
     AtomicInteger refreshCount = new AtomicInteger();
 
     ConnectionInfoRepository connectionInfoRepository =
-        (instanceName, accessTokenSupplier, authType, executor, keyPair) -> {
-          refreshCount.incrementAndGet();
-          return Futures.immediateFuture(info);
+        new StubConnectionInfoRepository() {
+          @Override
+          public ListenableFuture<ConnectionInfo> getConnectionInfo(
+              CloudSqlInstanceName instanceName,
+              AccessTokenSupplier accessTokenSupplier,
+              AuthType authType,
+              ListeningScheduledExecutorService executor,
+              ListenableFuture<KeyPair> keyPair) {
+            refreshCount.incrementAndGet();
+            return Futures.immediateFuture(info);
+          }
         };
 
     // initialize connectionInfoCache after mocks are set up
@@ -613,17 +695,27 @@ public class DefaultConnectionInfoCacheTest {
     AtomicInteger refreshCount = new AtomicInteger();
     final PauseCondition refresh0 = new PauseCondition();
 
+    ConnectionInfoRepository connectionInfoRepository =
+        new StubConnectionInfoRepository() {
+          @Override
+          public ListenableFuture<ConnectionInfo> getConnectionInfo(
+              CloudSqlInstanceName instanceName,
+              AccessTokenSupplier accessTokenSupplier,
+              AuthType authType,
+              ListeningScheduledExecutorService executor,
+              ListenableFuture<KeyPair> keyPair) {
+            int c = refreshCount.get();
+            if (c == 0) {
+              refresh0.pause();
+            }
+            refreshCount.incrementAndGet();
+            return Futures.immediateFuture(initialData);
+          }
+        };
     DefaultConnectionInfoCache instance =
         new DefaultConnectionInfoCache(
             new ConnectionConfig.Builder().withCloudSqlInstance("project:region:instance").build(),
-            (instanceName, accessTokenSupplier, authType, executor, keyPair) -> {
-              int c = refreshCount.get();
-              if (c == 0) {
-                refresh0.pause();
-              }
-              refreshCount.incrementAndGet();
-              return Futures.immediateFuture(initialData);
-            },
+            connectionInfoRepository,
             stubCredentialFactory,
             executorService,
             keyPairFuture,
diff --git a/core/src/test/java/com/google/cloud/sql/core/StubConnectionInfoRepository.java b/core/src/test/java/com/google/cloud/sql/core/StubConnectionInfoRepository.java
index dd6eb701..52b62790 100644
--- a/core/src/test/java/com/google/cloud/sql/core/StubConnectionInfoRepository.java
+++ b/core/src/test/java/com/google/cloud/sql/core/StubConnectionInfoRepository.java
@@ -31,7 +31,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import javax.net.ssl.KeyManagerFactory;
 
 class StubConnectionInfoRepository implements ConnectionInfoRepository {
-  private AtomicInteger refreshCount = new AtomicInteger();
+  private final AtomicInteger refreshCount = new AtomicInteger();
 
   int getRefreshCount() {
     return refreshCount.get();
@@ -61,4 +61,14 @@ class StubConnectionInfoRepository implements ConnectionInfoRepository {
     ConnectionInfo connectionInfo = newConnectionInfo();
     return Futures.immediateFuture(connectionInfo);
   }
+
+  @Override
+  public ConnectionInfo getConnectionInfoSync(
+      CloudSqlInstanceName instanceName,
+      AccessTokenSupplier accessTokenSupplier,
+      AuthType authType,
+      KeyPair keyPair) {
+    refreshCount.incrementAndGet();
+    return newConnectionInfo();
+  }
 }
diff --git a/core/src/test/java/com/google/cloud/sql/core/TestDataSupplier.java b/core/src/test/java/com/google/cloud/sql/core/TestDataSupplier.java
index 5b46d0c3..8a88079c 100644
--- a/core/src/test/java/com/google/cloud/sql/core/TestDataSupplier.java
+++ b/core/src/test/java/com/google/cloud/sql/core/TestDataSupplier.java
@@ -86,4 +86,18 @@ class TestDataSupplier implements ConnectionInfoRepository {
 
     return f;
   }
+
+  @Override
+  public ConnectionInfo getConnectionInfoSync(
+      CloudSqlInstanceName instanceName,
+      AccessTokenSupplier accessTokenSupplier,
+      AuthType authType,
+      KeyPair keyPair) {
+    int c = counter.incrementAndGet();
+    if (flaky && c % 2 == 0) {
+      throw new RuntimeException("Flaky");
+    }
+    successCounter.incrementAndGet();
+    return null;
+  }
 }