chore: use ScheduledExecutorService instead of Executor

BenWhitehead · BenWhitehead · commit 8553cb9d6153 · 2025-03-14T16:47:39.000-04:00
pre-work

For retries we will need the ability to schedule work in the future. Rather than sleeping on the current thread we can schedule the run of an operation after the backoff duration with ScheduledExecutorService#schedule.

Sourcing the executor service will first attempt to get it from the storage client settings, if it is unable to be resolved (either beacuse settings are null, or the returned provider) construct our own.
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobDescriptorImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobDescriptorImpl.java
@@ -33,7 +33,7 @@
 import io.grpc.Status.Code;
 import java.io.IOException;
 import java.util.OptionalLong;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
 
 final class BlobDescriptorImpl implements BlobDescriptor {
 
@@ -115,7 +115,7 @@ static ApiFuture<BlobDescriptor> create(
       BidiReadObjectRequest openRequest,
       GrpcCallContext context,
       ZeroCopyBidiStreamingCallable<BidiReadObjectRequest, BidiReadObjectResponse> callable,
-      Executor executor) {
+      ScheduledExecutorService executor) {
     BlobDescriptorState state = new BlobDescriptorState(openRequest);
 
     BlobDescriptorStream stream = BlobDescriptorStream.create(executor, callable, context, state);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobDescriptorStream.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobDescriptorStream.java
@@ -42,6 +42,7 @@
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -52,7 +53,7 @@ final class BlobDescriptorStream
   private final SettableApiFuture<Void> blobDescriptorResolveFuture;
 
   private final BlobDescriptorState state;
-  private final Executor executor;
+  private final ScheduledExecutorService executor;
   private final ZeroCopyBidiStreamingCallable<BidiReadObjectRequest, BidiReadObjectResponse>
       callable;
   private final GrpcCallContext context;
@@ -67,7 +68,7 @@ final class BlobDescriptorStream
 
   private BlobDescriptorStream(
       BlobDescriptorState state,
-      Executor executor,
+      ScheduledExecutorService executor,
       ZeroCopyBidiStreamingCallable<BidiReadObjectRequest, BidiReadObjectResponse> callable,
       GrpcCallContext context,
       int maxRedirectsAllowed) {
@@ -452,7 +453,7 @@ public void onComplete() {
   }
 
   static BlobDescriptorStream create(
-      Executor executor,
+      ScheduledExecutorService executor,
       ZeroCopyBidiStreamingCallable<BidiReadObjectRequest, BidiReadObjectResponse> callable,
       GrpcCallContext context,
       BlobDescriptorState state) {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -32,6 +32,7 @@
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
 import com.google.api.core.BetaApi;
+import com.google.api.gax.core.InstantiatingExecutorProvider;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.paging.AbstractPage;
 import com.google.api.gax.paging.Page;
@@ -105,6 +106,7 @@
 import com.google.storage.v2.RewriteObjectRequest;
 import com.google.storage.v2.RewriteResponse;
 import com.google.storage.v2.StorageClient;
+import com.google.storage.v2.StorageSettings;
 import com.google.storage.v2.UpdateBucketRequest;
 import com.google.storage.v2.UpdateObjectRequest;
 import com.google.storage.v2.WriteObjectRequest;
@@ -134,8 +136,7 @@
 import java.util.Spliterators.AbstractSpliterator;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executor;
-import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
@@ -175,7 +176,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
   final GrpcConversions codecs;
   final GrpcRetryAlgorithmManager retryAlgorithmManager;
   final SyntaxDecoders syntaxDecoders;
-  final Executor executor;
+  final ScheduledExecutorService executor;
 
   // workaround for https://github.com/googleapis/java-storage/issues/1736
   private final Opts<UserProject> defaultOpts;
@@ -198,7 +199,24 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     this.retryAlgorithmManager = options.getRetryAlgorithmManager();
     this.syntaxDecoders = new SyntaxDecoders();
     this.defaultProjectId = Suppliers.memoize(() -> UnifiedOpts.projectId(options.getProjectId()));
-    this.executor = Executors.newCachedThreadPool();
+    this.executor =
+        Utils.firstNonNull(
+                () -> {
+                  if (storageClient == null) {
+                    return null;
+                  }
+                  StorageSettings settings = storageClient.getSettings();
+                  if (settings == null) {
+                    return null;
+                  }
+                  return settings.getBackgroundExecutorProvider();
+                },
+                () -> {
+                  // TODO: if we make it to here, ensure we track the need to shutdown the executor
+                  //   separate from StorageClient
+                  return InstantiatingExecutorProvider.newBuilder().build();
+                })
+            .getExecutor();
   }
 
   @Override
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -42,6 +42,7 @@
 import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.api.gax.rpc.StatusCode.Code;
 import com.google.api.gax.rpc.internal.QuotaProjectIdHidingCredentials;
+import com.google.api.gax.tracing.ApiTracerFactory;
 import com.google.api.pathtemplate.PathTemplate;
 import com.google.auth.Credentials;
 import com.google.cloud.NoCredentials;
@@ -651,6 +652,20 @@ public GrpcStorageOptions.Builder setBlobWriteSessionConfig(
       return this;
     }
 
+    @BetaApi
+    @Override
+    public GrpcStorageOptions.Builder setUniverseDomain(String universeDomain) {
+      super.setUniverseDomain(universeDomain);
+      return this;
+    }
+
+    @BetaApi
+    @Override
+    public GrpcStorageOptions.Builder setApiTracerFactory(ApiTracerFactory apiTracerFactory) {
+      super.setApiTracerFactory(apiTracerFactory);
+      return this;
+    }
+
     /**
      * Enable OpenTelemetry Tracing and provide an instance for the client to use.
      *
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
@@ -24,6 +24,7 @@
 import com.google.api.core.InternalApi;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.HeaderProvider;
+import com.google.api.gax.tracing.ApiTracerFactory;
 import com.google.auth.Credentials;
 import com.google.cloud.ServiceFactory;
 import com.google.cloud.ServiceRpc;
@@ -270,6 +271,18 @@ public HttpStorageOptions.Builder setBlobWriteSessionConfig(
       return this;
     }
 
+    @Override
+    public HttpStorageOptions.Builder setUniverseDomain(String universeDomain) {
+      super.setUniverseDomain(universeDomain);
+      return this;
+    }
+
+    @Override
+    public HttpStorageOptions.Builder setApiTracerFactory(ApiTracerFactory apiTracerFactory) {
+      super.setApiTracerFactory(apiTracerFactory);
+      return this;
+    }
+
     @Override
     public HttpStorageOptions build() {
       HttpStorageOptions options = new HttpStorageOptions(this, defaults());
