FaultTolerantHttpClient: used managed ScheduledExecutorService for retries

integration-tests/src/main/java/org/signal/integration/Operations.java

@@ -311,6 +311,7 @@ public final class Operations {
       return FaultTolerantHttpClient.newBuilder()
           .withName("integration-test")
           .withExecutor(Executors.newFixedThreadPool(16))
+          .withRetryExecutor(Executors.newSingleThreadScheduledExecutor())
           .withCircuitBreaker(new CircuitBreakerConfiguration())
           .withTrustedServerCertificates(CONFIG.rootCert())
           .build();

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java

@@ -379,6 +379,10 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         .executorService(name(getClass(), "secureValueRecoveryService-%d")).maxThreads(1).minThreads(1).build();
     ExecutorService storageServiceExecutor = environment.lifecycle()
         .executorService(name(getClass(), "storageService-%d")).maxThreads(1).minThreads(1).build();
+    ScheduledExecutorService secureValueRecoveryServiceRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "secureValueRecoveryServiceRetry-%d")).threads(1).build();
+    ScheduledExecutorService storageServiceRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "storageServiceRetry-%d")).threads(1).build();
 
     Scheduler messageDeliveryScheduler = Schedulers.fromExecutorService(
         ExecutorServiceMetrics.monitor(Metrics.globalRegistry,
@@ -411,6 +415,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         .maxThreads(2)
         .minThreads(2)
         .build();
+    ScheduledExecutorService subscriptionProcessorRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "subscriptionProcessorRetry-%d")).threads(1).build();
 
     final AdminEventLogger adminEventLogger = new GoogleCloudAdminEventLogger(
         LoggingOptions.newBuilder().setProjectId(config.getAdminEventLoggingConfiguration().projectId())
@@ -426,9 +432,11 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         config.getStripe().idempotencyKeyGenerator().value(), config.getStripe().boostDescription(), config.getStripe()
         .supportedCurrencies());
     BraintreeManager braintreeManager = new BraintreeManager(config.getBraintree().merchantId(),
-        config.getBraintree().publicKey(), config.getBraintree().privateKey().value(), config.getBraintree().environment(),
+        config.getBraintree().publicKey(), config.getBraintree().privateKey().value(),
+        config.getBraintree().environment(),
         config.getBraintree().supportedCurrencies(), config.getBraintree().merchantAccounts(),
-        config.getBraintree().graphqlUrl(), config.getBraintree().circuitBreaker(), subscriptionProcessorExecutor);
+        config.getBraintree().graphqlUrl(), config.getBraintree().circuitBreaker(), subscriptionProcessorExecutor,
+        subscriptionProcessorRetryExecutor);
 
     ExternalServiceCredentialsGenerator directoryV2CredentialsGenerator = DirectoryV2Controller.credentialsGenerator(
         config.getDirectoryV2Configuration().getDirectoryV2ClientConfiguration());
@@ -461,11 +469,12 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         config.getRegistrationServiceConfiguration().registrationCaCertificate(),
         registrationCallbackExecutor);
     SecureBackupClient secureBackupClient = new SecureBackupClient(backupCredentialsGenerator,
-        secureValueRecoveryServiceExecutor, config.getSecureBackupServiceConfiguration());
+        secureValueRecoveryServiceExecutor, secureValueRecoveryServiceRetryExecutor,
+        config.getSecureBackupServiceConfiguration());
     SecureValueRecovery2Client secureValueRecovery2Client = new SecureValueRecovery2Client(svr2CredentialsGenerator,
-        secureValueRecoveryServiceExecutor, config.getSvr2Configuration());
+        secureValueRecoveryServiceExecutor, secureValueRecoveryServiceRetryExecutor, config.getSvr2Configuration());
     SecureStorageClient secureStorageClient = new SecureStorageClient(storageCredentialsGenerator,
-        storageServiceExecutor, config.getSecureStorageServiceConfiguration());
+        storageServiceExecutor, storageServiceRetryExecutor, config.getSecureStorageServiceConfiguration());
     ClientPresenceManager clientPresenceManager = new ClientPresenceManager(clientPresenceCluster, recurringJobExecutor,
         keyspaceNotificationDispatchExecutor);
     StoredVerificationCodeManager pendingAccountsManager = new StoredVerificationCodeManager(pendingAccounts);

service/src/main/java/org/whispersystems/textsecuregcm/http/FaultTolerantHttpClient.java

@@ -17,7 +17,6 @@ import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Executor;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Supplier;
 import org.glassfish.jersey.SslConfigurator;
@@ -40,15 +39,20 @@ public class FaultTolerantHttpClient {
     return new Builder();
   }
 
-  private FaultTolerantHttpClient(String name, HttpClient httpClient, RetryConfiguration retryConfiguration, CircuitBreakerConfiguration circuitBreakerConfiguration) {
+  private FaultTolerantHttpClient(String name, HttpClient httpClient, ScheduledExecutorService retryExecutor,
+      RetryConfiguration retryConfiguration, CircuitBreakerConfiguration circuitBreakerConfiguration) {
     this.httpClient = httpClient;
-    this.retryExecutor = Executors.newSingleThreadScheduledExecutor();
+    this.retryExecutor = retryExecutor;
     this.breaker = CircuitBreaker.of(name + "-breaker", circuitBreakerConfiguration.toCircuitBreakerConfig());
 
     CircuitBreakerUtil.registerMetrics(breaker, FaultTolerantHttpClient.class);
 
     if (retryConfiguration != null) {
-      RetryConfig retryConfig = retryConfiguration.<HttpResponse>toRetryConfigBuilder().retryOnResult(o -> o.statusCode() >= 500).build();
+      if (this.retryExecutor == null) {
+        throw new IllegalArgumentException("retryExecutor must be specified with retryConfiguration");
+      }
+      RetryConfig retryConfig = retryConfiguration.<HttpResponse>toRetryConfigBuilder()
+          .retryOnResult(o -> o.statusCode() >= 500).build();
       this.retry = Retry.of(name + "-retry", retryConfig);
       CircuitBreakerUtil.registerMetrics(retry, FaultTolerantHttpClient.class);
     } else {
@@ -76,19 +80,20 @@ public class FaultTolerantHttpClient {
 
   public static class Builder {
 
-
     private HttpClient.Version version = HttpClient.Version.HTTP_2;
     private HttpClient.Redirect redirect = HttpClient.Redirect.NEVER;
     private Duration connectTimeout = Duration.ofSeconds(10);
 
     private String name;
     private Executor executor;
+    private ScheduledExecutorService retryExecutor;
     private KeyStore trustStore;
     private String securityProtocol = SECURITY_PROTOCOL_TLS_1_2;
     private RetryConfiguration retryConfiguration;
     private CircuitBreakerConfiguration circuitBreakerConfiguration;
 
-    private Builder() {}
+    private Builder() {
+    }
 
     public Builder withName(String name) {
       this.name = name;
@@ -120,6 +125,11 @@ public class FaultTolerantHttpClient {
       return this;
     }
 
+    public Builder withRetryExecutor(ScheduledExecutorService retryExecutor) {
+      this.retryExecutor = retryExecutor;
+      return this;
+    }
+
     public Builder withCircuitBreaker(CircuitBreakerConfiguration circuitBreakerConfiguration) {
       this.circuitBreakerConfiguration = circuitBreakerConfiguration;
       return this;
@@ -154,9 +164,9 @@ public class FaultTolerantHttpClient {
 
       builder.sslContext(sslConfigurator.createSSLContext());
 
-      return new FaultTolerantHttpClient(name, builder.build(), retryConfiguration, circuitBreakerConfiguration);
+      return new FaultTolerantHttpClient(name, builder.build(), retryExecutor, retryConfiguration,
+          circuitBreakerConfiguration);
     }
-
   }
 
 }

service/src/main/java/org/whispersystems/textsecuregcm/securebackup/SecureBackupClient.java

@@ -18,6 +18,7 @@ import java.time.Duration;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentials;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialsGenerator;
 import org.whispersystems.textsecuregcm.configuration.SecureBackupServiceConfiguration;
@@ -36,12 +37,16 @@ public class SecureBackupClient {
   @VisibleForTesting
   static final String DELETE_PATH = "/v1/backup";
 
-    public SecureBackupClient(final ExternalServiceCredentialsGenerator secureBackupCredentialsGenerator, final Executor executor, final SecureBackupServiceConfiguration configuration) throws CertificateException {
+  public SecureBackupClient(final ExternalServiceCredentialsGenerator secureBackupCredentialsGenerator,
+      final Executor executor, final
+  ScheduledExecutorService retryExecutor, final SecureBackupServiceConfiguration configuration)
+      throws CertificateException {
     this.secureBackupCredentialsGenerator = secureBackupCredentialsGenerator;
     this.deleteUri = URI.create(configuration.getUri()).resolve(DELETE_PATH);
     this.httpClient = FaultTolerantHttpClient.newBuilder()
         .withCircuitBreaker(configuration.getCircuitBreakerConfiguration())
         .withRetry(configuration.getRetryConfiguration())
+        .withRetryExecutor(retryExecutor)
         .withVersion(HttpClient.Version.HTTP_1_1)
         .withConnectTimeout(Duration.ofSeconds(10))
         .withRedirect(HttpClient.Redirect.NEVER)

service/src/main/java/org/whispersystems/textsecuregcm/securestorage/SecureStorageClient.java

@@ -18,6 +18,7 @@ import java.time.Duration;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentials;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialsGenerator;
 import org.whispersystems.textsecuregcm.configuration.SecureStorageServiceConfiguration;
@@ -36,12 +37,16 @@ public class SecureStorageClient {
   @VisibleForTesting
   static final String DELETE_PATH = "/v1/storage";
 
-    public SecureStorageClient(final ExternalServiceCredentialsGenerator storageServiceCredentialsGenerator, final Executor executor, final SecureStorageServiceConfiguration configuration) throws CertificateException {
+  public SecureStorageClient(final ExternalServiceCredentialsGenerator storageServiceCredentialsGenerator,
+      final Executor executor, final
+  ScheduledExecutorService retryExecutor, final SecureStorageServiceConfiguration configuration)
+      throws CertificateException {
     this.storageServiceCredentialsGenerator = storageServiceCredentialsGenerator;
     this.deleteUri = URI.create(configuration.uri()).resolve(DELETE_PATH);
     this.httpClient = FaultTolerantHttpClient.newBuilder()
         .withCircuitBreaker(configuration.circuitBreaker())
         .withRetry(configuration.retry())
+        .withRetryExecutor(retryExecutor)
         .withVersion(HttpClient.Version.HTTP_1_1)
         .withConnectTimeout(Duration.ofSeconds(10))
         .withRedirect(HttpClient.Redirect.NEVER)

service/src/main/java/org/whispersystems/textsecuregcm/securevaluerecovery/SecureValueRecovery2Client.java

@@ -18,6 +18,7 @@ import java.time.Duration;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentials;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialsGenerator;
 import org.whispersystems.textsecuregcm.configuration.SecureValueRecovery2Configuration;
@@ -37,13 +38,15 @@ public class SecureValueRecovery2Client {
   static final String DELETE_PATH = "/v1/delete";
 
   public SecureValueRecovery2Client(final ExternalServiceCredentialsGenerator secureValueRecoveryCredentialsGenerator,
-      final Executor executor, final SecureValueRecovery2Configuration configuration)
+      final Executor executor, final ScheduledExecutorService retryExecutor,
+      final SecureValueRecovery2Configuration configuration)
       throws CertificateException {
     this.secureValueRecoveryCredentialsGenerator = secureValueRecoveryCredentialsGenerator;
     this.deleteUri = URI.create(configuration.uri()).resolve(DELETE_PATH);
     this.httpClient = FaultTolerantHttpClient.newBuilder()
         .withCircuitBreaker(configuration.circuitBreaker())
         .withRetry(configuration.retry())
+        .withRetryExecutor(retryExecutor)
         .withVersion(HttpClient.Version.HTTP_1_1)
         .withConnectTimeout(Duration.ofSeconds(10))
         .withRedirect(HttpClient.Redirect.NEVER)

service/src/main/java/org/whispersystems/textsecuregcm/subscriptions/BraintreeManager.java

@@ -32,6 +32,7 @@ import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
 import javax.annotation.Nullable;
 import javax.ws.rs.ClientErrorException;
 import javax.ws.rs.WebApplicationException;
@@ -61,7 +62,8 @@ public class BraintreeManager implements SubscriptionProcessorManager {
       final Map<String, String> currenciesToMerchantAccounts,
       final String graphqlUri,
       final CircuitBreakerConfiguration circuitBreakerConfiguration,
-      final Executor executor) {
+      final Executor executor,
+      final ScheduledExecutorService retryExecutor) {
 
     this(new BraintreeGateway(braintreeEnvironment, braintreeMerchantId, braintreePublicKey,
             braintreePrivateKey),
@@ -71,6 +73,7 @@ public class BraintreeManager implements SubscriptionProcessorManager {
             .withName("braintree-graphql")
             .withCircuitBreaker(circuitBreakerConfiguration)
             .withExecutor(executor)
+            .withRetryExecutor(retryExecutor)
             .build(), graphqlUri, braintreePublicKey, braintreePrivateKey),
         executor);
   }

service/src/main/java/org/whispersystems/textsecuregcm/workers/AssignUsernameCommand.java

@@ -18,6 +18,7 @@ import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import net.sourceforge.argparse4j.inf.Namespace;
 import net.sourceforge.argparse4j.inf.Subparser;
 import org.whispersystems.textsecuregcm.WhisperServerConfiguration;
@@ -110,6 +111,10 @@ public class AssignUsernameCommand extends EnvironmentCommand<WhisperServerConfi
         .executorService(name(getClass(), "secureValueRecoveryService-%d")).maxThreads(1).minThreads(1).build();
     ExecutorService storageServiceExecutor = environment.lifecycle()
         .executorService(name(getClass(), "storageService-%d")).maxThreads(1).minThreads(1).build();
+    ScheduledExecutorService secureValueRecoveryServiceRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "secureValueRecoveryServiceRetry-%d")).threads(1).build();
+    ScheduledExecutorService storageServiceRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "storageServiceRetry-%d")).threads(1).build();
 
     ExternalServiceCredentialsGenerator backupCredentialsGenerator = SecureBackupController.credentialsGenerator(
         configuration.getSecureBackupServiceConfiguration());
@@ -174,12 +179,14 @@ public class AssignUsernameCommand extends EnvironmentCommand<WhisperServerConfi
         configuration.getClientPresenceClusterConfiguration(), redisClusterClientResources);
     FaultTolerantRedisCluster rateLimitersCluster = new FaultTolerantRedisCluster("rate_limiters",
         configuration.getRateLimitersCluster(), redisClusterClientResources);
-    SecureBackupClient secureBackupClient = new SecureBackupClient(backupCredentialsGenerator, secureValueRecoveryExecutor,
+    SecureBackupClient secureBackupClient = new SecureBackupClient(backupCredentialsGenerator,
-        configuration.getSecureBackupServiceConfiguration());
+        secureValueRecoveryExecutor,
+        secureValueRecoveryServiceRetryExecutor, configuration.getSecureBackupServiceConfiguration());
     SecureValueRecovery2Client secureValueRecovery2Client = new SecureValueRecovery2Client(
-        secureValueRecoveryCredentialsGenerator, secureValueRecoveryExecutor, configuration.getSvr2Configuration());
+        secureValueRecoveryCredentialsGenerator, secureValueRecoveryExecutor, secureValueRecoveryServiceRetryExecutor,
+        configuration.getSvr2Configuration());
     SecureStorageClient secureStorageClient = new SecureStorageClient(storageCredentialsGenerator,
-        storageServiceExecutor, configuration.getSecureStorageServiceConfiguration());
+        storageServiceExecutor, storageServiceRetryExecutor, configuration.getSecureStorageServiceConfiguration());
     ClientPresenceManager clientPresenceManager = new ClientPresenceManager(clientPresenceCluster,
         Executors.newSingleThreadScheduledExecutor(), keyspaceNotificationDispatchExecutor);
     MessagesCache messagesCache = new MessagesCache(messageInsertCacheCluster, messageReadDeleteCluster,

service/src/main/java/org/whispersystems/textsecuregcm/workers/CommandDependencies.java

@@ -94,6 +94,11 @@ record CommandDependencies(
     ExecutorService storageServiceExecutor = environment.lifecycle()
         .executorService(name(name, "storageService-%d")).maxThreads(8).minThreads(8).build();
 
+    ScheduledExecutorService secureValueRecoveryServiceRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(name, "secureValueRecoveryServiceRetry-%d")).threads(1).build();
+    ScheduledExecutorService storageServiceRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(name, "storageServiceRetry-%d")).threads(1).build();
+
     ExternalServiceCredentialsGenerator backupCredentialsGenerator = SecureBackupController.credentialsGenerator(
         configuration.getSecureBackupServiceConfiguration());
     ExternalServiceCredentialsGenerator storageCredentialsGenerator = SecureStorageController.credentialsGenerator(
@@ -159,13 +164,14 @@ record CommandDependencies(
     FaultTolerantRedisCluster rateLimitersCluster = new FaultTolerantRedisCluster("rate_limiters",
         configuration.getRateLimitersCluster(), redisClusterClientResources);
     SecureBackupClient secureBackupClient = new SecureBackupClient(backupCredentialsGenerator,
-        secureValueRecoveryServiceExecutor,
+        secureValueRecoveryServiceExecutor, secureValueRecoveryServiceRetryExecutor,
         configuration.getSecureBackupServiceConfiguration());
     SecureValueRecovery2Client secureValueRecovery2Client = new SecureValueRecovery2Client(
         secureValueRecoveryCredentialsGenerator, secureValueRecoveryServiceExecutor,
+        secureValueRecoveryServiceRetryExecutor,
         configuration.getSvr2Configuration());
     SecureStorageClient secureStorageClient = new SecureStorageClient(storageCredentialsGenerator,
-        storageServiceExecutor, configuration.getSecureStorageServiceConfiguration());
+        storageServiceExecutor, storageServiceRetryExecutor, configuration.getSecureStorageServiceConfiguration());
     ClientPresenceManager clientPresenceManager = new ClientPresenceManager(clientPresenceCluster,
         recurringJobExecutor, keyspaceNotificationDispatchExecutor);
     MessagesCache messagesCache = new MessagesCache(messageInsertCacheCluster, messageReadDeleteCluster,

service/src/test/java/org/whispersystems/textsecuregcm/securebackup/SecureBackupClientTest.java

@@ -21,6 +21,7 @@ import java.util.UUID;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.AfterEach;
@@ -36,6 +37,7 @@ class SecureBackupClientTest {
   private UUID accountUuid;
   private ExternalServiceCredentialsGenerator credentialsGenerator;
   private ExecutorService httpExecutor;
+  private ScheduledExecutorService retryExecutor;
 
   private SecureBackupClient secureStorageClient;
 
@@ -49,12 +51,14 @@ class SecureBackupClientTest {
     accountUuid = UUID.randomUUID();
     credentialsGenerator = mock(ExternalServiceCredentialsGenerator.class);
     httpExecutor = Executors.newSingleThreadExecutor();
+    retryExecutor = Executors.newSingleThreadScheduledExecutor();
 
     final SecureBackupServiceConfiguration config = new SecureBackupServiceConfiguration();
     config.setUri("http://localhost:" + wireMock.getPort());
 
     // This is a randomly-generated, throwaway certificate that's not actually connected to anything
-    config.setBackupCaCertificates(List.of("""
+    config.setBackupCaCertificates(
+        List.of("""
             -----BEGIN CERTIFICATE-----
             MIICZDCCAc2gAwIBAgIBADANBgkqhkiG9w0BAQ0FADBPMQswCQYDVQQGEwJ1czEL
             MAkGA1UECAwCVVMxHjAcBgNVBAoMFVNpZ25hbCBNZXNzZW5nZXIsIExMQzETMBEG
@@ -70,8 +74,7 @@ class SecureBackupClientTest {
             y7MTM4NoBV1k0zb5LAk89SIDPr/maW5AsLtEomzjnEiomjoMBUdNe3YCgQReoLnr
             R/QaUNbrCjTGYfBsjGbIzmkWPUyTec2ZdRyJ8JiVl386+6CZkxnndQ==
             -----END CERTIFICATE-----
-        """,
+            """, """
-        """
             -----BEGIN CERTIFICATE-----
             MIIEpDCCAowCCQC43PUTWSADVjANBgkqhkiG9w0BAQsFADAUMRIwEAYDVQQDDAls
             b2NhbGhvc3QwHhcNMjIxMDE3MjA0NTM0WhcNMjMxMDE3MjA0NTM0WjAUMRIwEAYD
@@ -101,13 +104,15 @@ class SecureBackupClientTest {
             -----END CERTIFICATE-----
             """));
 
-    secureStorageClient = new SecureBackupClient(credentialsGenerator, httpExecutor, config);
+    secureStorageClient = new SecureBackupClient(credentialsGenerator, httpExecutor, retryExecutor, config);
   }
 
   @AfterEach
   void tearDown() throws InterruptedException {
     httpExecutor.shutdown();
     httpExecutor.awaitTermination(1, TimeUnit.SECONDS);
+    retryExecutor.shutdown();
+    retryExecutor.awaitTermination(1, TimeUnit.SECONDS);
   }
 
   @Test

service/src/test/java/org/whispersystems/textsecuregcm/securestorage/SecureStorageClientTest.java

@@ -22,6 +22,7 @@ import java.util.UUID;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.AfterEach;
@@ -39,6 +40,7 @@ class SecureStorageClientTest {
   private UUID accountUuid;
   private ExternalServiceCredentialsGenerator credentialsGenerator;
   private ExecutorService httpExecutor;
+  private ScheduledExecutorService retryExecutor;
 
   private SecureStorageClient secureStorageClient;
 
@@ -52,6 +54,7 @@ class SecureStorageClientTest {
     accountUuid = UUID.randomUUID();
     credentialsGenerator = mock(ExternalServiceCredentialsGenerator.class);
     httpExecutor = Executors.newSingleThreadExecutor();
+    retryExecutor = Executors.newSingleThreadScheduledExecutor();
 
     final SecureStorageServiceConfiguration config = new SecureStorageServiceConfiguration(
         randomSecretBytes(32),
@@ -72,8 +75,7 @@ class SecureStorageClientTest {
             y7MTM4NoBV1k0zb5LAk89SIDPr/maW5AsLtEomzjnEiomjoMBUdNe3YCgQReoLnr
             R/QaUNbrCjTGYfBsjGbIzmkWPUyTec2ZdRyJ8JiVl386+6CZkxnndQ==
             -----END CERTIFICATE-----
-                """,
+            """, """
-                """
             -----BEGIN CERTIFICATE-----
             MIIEpDCCAowCCQC43PUTWSADVjANBgkqhkiG9w0BAQsFADAUMRIwEAYDVQQDDAls
             b2NhbGhvc3QwHhcNMjIxMDE3MjA0NTM0WhcNMjMxMDE3MjA0NTM0WjAUMRIwEAYD
@@ -105,21 +107,26 @@ class SecureStorageClientTest {
         new CircuitBreakerConfiguration(),
         new RetryConfiguration());
 
-        secureStorageClient = new SecureStorageClient(credentialsGenerator, httpExecutor, config);
+    secureStorageClient = new SecureStorageClient(credentialsGenerator, httpExecutor, retryExecutor, config);
   }
 
   @AfterEach
   void tearDown() throws InterruptedException {
+
     httpExecutor.shutdown();
     httpExecutor.awaitTermination(1, TimeUnit.SECONDS);
+    retryExecutor.shutdown();
+    retryExecutor.awaitTermination(1, TimeUnit.SECONDS);
   }
 
   @Test
   void deleteStoredData() {
+
     final String username = RandomStringUtils.randomAlphabetic(16);
     final String password = RandomStringUtils.randomAlphanumeric(32);
 
-        when(credentialsGenerator.generateForUuid(accountUuid)).thenReturn(new ExternalServiceCredentials(username, password));
+    when(credentialsGenerator.generateForUuid(accountUuid)).thenReturn(
+        new ExternalServiceCredentials(username, password));
 
     wireMock.stubFor(delete(urlEqualTo(SecureStorageClient.DELETE_PATH))
         .withBasicAuth(username, password)
@@ -131,16 +138,19 @@ class SecureStorageClientTest {
 
   @Test
   void deleteStoredDataFailure() {
+
     final String username = RandomStringUtils.randomAlphabetic(16);
     final String password = RandomStringUtils.randomAlphanumeric(32);
 
-        when(credentialsGenerator.generateForUuid(accountUuid)).thenReturn(new ExternalServiceCredentials(username, password));
+    when(credentialsGenerator.generateForUuid(accountUuid)).thenReturn(
+        new ExternalServiceCredentials(username, password));
 
     wireMock.stubFor(delete(urlEqualTo(SecureStorageClient.DELETE_PATH))
         .withBasicAuth(username, password)
         .willReturn(aResponse().withStatus(400)));
 
-        final CompletionException completionException = assertThrows(CompletionException.class, () -> secureStorageClient.deleteStoredData(accountUuid).join());
+    final CompletionException completionException = assertThrows(CompletionException.class,
+        () -> secureStorageClient.deleteStoredData(accountUuid).join());
     assertTrue(completionException.getCause() instanceof SecureStorageException);
   }
 }

service/src/test/java/org/whispersystems/textsecuregcm/securevaluerecovery/SecureValueRecovery2ClientTest.java

@@ -23,6 +23,7 @@ import java.util.UUID;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.AfterEach;
@@ -38,6 +39,7 @@ class SecureValueRecovery2ClientTest {
   private UUID accountUuid;
   private ExternalServiceCredentialsGenerator credentialsGenerator;
   private ExecutorService httpExecutor;
+  private ScheduledExecutorService retryExecutor;
 
   private SecureValueRecovery2Client secureValueRecovery2Client;
 
@@ -51,6 +53,7 @@ class SecureValueRecovery2ClientTest {
     accountUuid = UUID.randomUUID();
     credentialsGenerator = mock(ExternalServiceCredentialsGenerator.class);
     httpExecutor = Executors.newSingleThreadExecutor();
+    retryExecutor = Executors.newSingleThreadScheduledExecutor();
 
     final SecureValueRecovery2Configuration config = new SecureValueRecovery2Configuration(
         "http://localhost:" + wireMock.getPort(),
@@ -104,13 +107,16 @@ class SecureValueRecovery2ClientTest {
             """),
         null, null);
 
-    secureValueRecovery2Client = new SecureValueRecovery2Client(credentialsGenerator, httpExecutor, config);
+    secureValueRecovery2Client = new SecureValueRecovery2Client(credentialsGenerator, httpExecutor, retryExecutor,
+        config);
   }
 
   @AfterEach
   void tearDown() throws InterruptedException {
     httpExecutor.shutdown();
     httpExecutor.awaitTermination(1, TimeUnit.SECONDS);
+    retryExecutor.shutdown();
+    retryExecutor.awaitTermination(1, TimeUnit.SECONDS);
   }
 
   @Test

service/src/test/java/org/whispersystems/textsecuregcm/tests/http/FaultTolerantHttpClientTest.java

@@ -20,7 +20,12 @@ import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.whispersystems.textsecuregcm.configuration.CircuitBreakerConfiguration;
@@ -34,6 +39,23 @@ class FaultTolerantHttpClientTest {
       .options(wireMockConfig().dynamicPort().dynamicHttpsPort())
       .build();
 
+  private ExecutorService httpExecutor;
+  private ScheduledExecutorService retryExecutor;
+
+  @BeforeEach
+  void setUp() {
+    httpExecutor = Executors.newSingleThreadExecutor();
+    retryExecutor = Executors.newSingleThreadScheduledExecutor();
+  }
+
+  @AfterEach
+  void tearDown() throws InterruptedException {
+    httpExecutor.shutdown();
+    httpExecutor.awaitTermination(1, TimeUnit.SECONDS);
+    retryExecutor.shutdown();
+    retryExecutor.awaitTermination(1, TimeUnit.SECONDS);
+  }
+
   @Test
   void testSimpleGet() {
     wireMock.stubFor(get(urlEqualTo("/ping"))
@@ -41,11 +63,11 @@ class FaultTolerantHttpClientTest {
             .withHeader("Content-Type", "text/plain")
             .withBody("Pong!")));
 
-
     FaultTolerantHttpClient client = FaultTolerantHttpClient.newBuilder()
         .withCircuitBreaker(new CircuitBreakerConfiguration())
         .withRetry(new RetryConfiguration())
-                                                            .withExecutor(Executors.newSingleThreadExecutor())
+        .withExecutor(httpExecutor)
+        .withRetryExecutor(retryExecutor)
         .withName("test")
         .withVersion(HttpClient.Version.HTTP_2)
         .build();
@@ -74,7 +96,8 @@ class FaultTolerantHttpClientTest {
     FaultTolerantHttpClient client = FaultTolerantHttpClient.newBuilder()
         .withCircuitBreaker(new CircuitBreakerConfiguration())
         .withRetry(new RetryConfiguration())
-                                                            .withExecutor(Executors.newSingleThreadExecutor())
+        .withExecutor(httpExecutor)
+        .withRetryExecutor(retryExecutor)
         .withName("test")
         .withVersion(HttpClient.Version.HTTP_2)
         .build();
@@ -104,7 +127,8 @@ class FaultTolerantHttpClientTest {
     FaultTolerantHttpClient client = FaultTolerantHttpClient.newBuilder()
         .withCircuitBreaker(circuitBreakerConfiguration)
         .withRetry(new RetryConfiguration())
-        .withExecutor(Executors.newSingleThreadExecutor())
+        .withRetryExecutor(retryExecutor)
+        .withExecutor(httpExecutor)
         .withName("test")
         .withVersion(HttpClient.Version.HTTP_2)
         .build();