fix: stop overriding the default gRPC executor

gax-grpc/src/main/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProvider.java

@@ -34,7 +34,6 @@
 import com.google.api.core.InternalApi;
 import com.google.api.core.InternalExtensionOnly;
 import com.google.api.gax.core.ExecutorProvider;
-import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.api.gax.rpc.FixedHeaderProvider;
 import com.google.api.gax.rpc.HeaderProvider;
 import com.google.api.gax.rpc.TransportChannel;
@@ -50,6 +49,7 @@
 import io.grpc.alts.ComputeEngineChannelBuilder;
 import java.io.IOException;
 import java.util.Map;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
@@ -76,7 +76,7 @@ public final class InstantiatingGrpcChannelProvider implements TransportChannelP
   static final int MAX_POOL_SIZE = 1000;
 
   private final int processorCount;
-  private final ExecutorProvider executorProvider;
+  @Nullable private final Executor executor;
   private final HeaderProvider headerProvider;
   private final String endpoint;
   private final EnvironmentProvider envProvider;
@@ -95,7 +95,7 @@ public final class InstantiatingGrpcChannelProvider implements TransportChannelP
 
   private InstantiatingGrpcChannelProvider(Builder builder) {
     this.processorCount = builder.processorCount;
-    this.executorProvider = builder.executorProvider;
+    this.executor = builder.executor;
     this.headerProvider = builder.headerProvider;
     this.endpoint = builder.endpoint;
     this.envProvider = builder.envProvider;
@@ -113,12 +113,25 @@ private InstantiatingGrpcChannelProvider(Builder builder) {
 
   @Override
   public boolean needsExecutor() {
-    return executorProvider == null;
+    // Use the default gRPC executor
+    return false;
   }
 
+  /** @deprecated Please use {@link #withExecutor(Executor) } */
+  @Deprecated
   @Override
   public TransportChannelProvider withExecutor(ScheduledExecutorService executor) {
-    return toBuilder().setExecutorProvider(FixedExecutorProvider.create(executor)).build();
+    return withExecutor((Executor) executor);
+  }
+
+  /**
+   * Sets the executor to use when constructing a new {@link TransportChannel}..
+   *
+   * <p>This can be used to override the default gRPC executor.
+   */
+  @Override
+  public TransportChannelProvider withExecutor(@Nullable Executor executor) {
+    return toBuilder().setExecutor(executor).build();
   }
 
   @Override
@@ -223,7 +236,6 @@ private boolean isDirectPathEnabled(String serviceAddress) {
   }
 
   private ManagedChannel createSingleChannel() throws IOException {
-    ScheduledExecutorService executor = executorProvider.getExecutor();
     GrpcHeaderInterceptor headerInterceptor =
         new GrpcHeaderInterceptor(headerProvider.getHeaders());
     GrpcMetadataHandlerInterceptor metadataHandlerInterceptor =
@@ -348,7 +360,7 @@ public static Builder newBuilder() {
 
   public static final class Builder {
     private int processorCount;
-    private ExecutorProvider executorProvider;
+    @Nullable private Executor executor;
     private HeaderProvider headerProvider;
     private String endpoint;
     private EnvironmentProvider envProvider;
@@ -370,7 +382,7 @@ private Builder() {
 
     private Builder(InstantiatingGrpcChannelProvider provider) {
       this.processorCount = provider.processorCount;
-      this.executorProvider = provider.executorProvider;
+      this.executor = provider.executor;
       this.headerProvider = provider.headerProvider;
       this.endpoint = provider.endpoint;
       this.envProvider = provider.envProvider;
@@ -398,16 +410,23 @@ Builder setEnvironmentProvider(EnvironmentProvider envProvider) {
       return this;
     }
 
+    /** @deprecated Please use {@link #setExecutor(Executor)} instead. */
+    @Deprecated
+    public Builder setExecutorProvider(ExecutorProvider executorProvider) {
+      Executor executor = null;
+      if (executorProvider != null) {
+        executor = executorProvider.getExecutor();
+      }
+      return setExecutor(executor);
+    }
+
     /**
-     * Sets the ExecutorProvider for this TransportChannelProvider.
+     * Sets the Executor for this TransportChannelProvider.
      *
-     * <p>This is optional; if it is not provided, needsExecutor() will return true, meaning that an
-     * Executor must be provided when getChannel is called on the constructed
-     * TransportChannelProvider instance. Note: GrpcTransportProvider will automatically provide its
-     * own Executor in this circumstance when it calls getChannel.
+     * <p>This is optional; if it is not set, the default gRPC executor will be used.
      */
-    public Builder setExecutorProvider(ExecutorProvider executorProvider) {
-      this.executorProvider = executorProvider;
+    public Builder setExecutor(@Nullable Executor executor) {
+      this.executor = executor;
       return this;
     }
 

gax-grpc/src/test/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProviderTest.java

@@ -36,19 +36,25 @@
 
 import com.google.api.core.ApiFunction;
 import com.google.api.gax.core.ExecutorProvider;
+import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.Builder;
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.EnvironmentProvider;
+import com.google.api.gax.grpc.testing.FakeServiceGrpc;
+import com.google.api.gax.rpc.FixedHeaderProvider;
 import com.google.api.gax.rpc.HeaderProvider;
 import com.google.api.gax.rpc.TransportChannelProvider;
 import com.google.auth.oauth2.CloudShellCredentials;
 import com.google.auth.oauth2.ComputeEngineCredentials;
-import io.grpc.ManagedChannel;
-import io.grpc.ManagedChannelBuilder;
+import com.google.common.util.concurrent.SettableFuture;
+import com.google.type.Color;
+import com.google.type.Money;
+import io.grpc.*;
 import io.grpc.alts.ComputeEngineChannelBuilder;
+import io.grpc.stub.ClientCalls;
+import io.grpc.stub.StreamObserver;
 import java.io.IOException;
 import java.util.Collections;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.*;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -385,4 +391,126 @@ public void testWithPrimeChannel() throws IOException {
           .primeChannel(Mockito.any(ManagedChannel.class));
     }
   }
+
+  @Test
+  public void testExecutorOverride() throws Exception {
+    final String expectedThreadName = "testExecutorOverrideExecutor";
+
+    ThreadPoolExecutor executor =
+        new ScheduledThreadPoolExecutor(
+            1,
+            new ThreadFactory() {
+              @Override
+              public Thread newThread(Runnable r) {
+                Thread thread = new Thread(r, expectedThreadName);
+                thread.setDaemon(true);
+                return thread;
+              }
+            });
+
+    try {
+      InstantiatingGrpcChannelProvider channelProvider =
+          InstantiatingGrpcChannelProvider.newBuilder()
+              // Don't actually care to connect to anything
+              .setEndpoint("localhost:1234")
+              .setHeaderProvider(FixedHeaderProvider.create())
+              .setExecutor(executor)
+              .build();
+
+      assertThat(extractExecutorThreadName(channelProvider)).isEqualTo(expectedThreadName);
+    } finally {
+      executor.shutdown();
+      executor.awaitTermination(10, TimeUnit.SECONDS);
+    }
+  }
+
+  @Test
+  @Deprecated
+  public void testDeprecatedExecutorOverride() throws Exception {
+    final String expectedThreadName = "testExecutorOverrideExecutor";
+
+    ScheduledExecutorService executor =
+        Executors.newScheduledThreadPool(
+            1,
+            new ThreadFactory() {
+              @Override
+              public Thread newThread(Runnable r) {
+                Thread thread = new Thread(r, expectedThreadName);
+                thread.setDaemon(true);
+                return thread;
+              }
+            });
+
+    try {
+      InstantiatingGrpcChannelProvider channelProvider =
+          InstantiatingGrpcChannelProvider.newBuilder()
+              // Don't actually care to connect to anything
+              .setEndpoint("localhost:1234")
+              .setHeaderProvider(FixedHeaderProvider.create())
+              .setExecutorProvider(FixedExecutorProvider.create(executor))
+              .build();
+
+      assertThat(extractExecutorThreadName(channelProvider)).isEqualTo(expectedThreadName);
+    } finally {
+      executor.shutdown();
+      executor.awaitTermination(10, TimeUnit.SECONDS);
+    }
+  }
+
+  @Test
+  public void testExecutorDefault() throws Exception {
+    InstantiatingGrpcChannelProvider channelProvider =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            // Don't actually care if we connect to anything
+            .setEndpoint("localhost:1234")
+            .setHeaderProvider(FixedHeaderProvider.create())
+            .build();
+
+    // The default name thread name for grpc threads configured in GrpcUtil
+    assertThat(extractExecutorThreadName(channelProvider)).contains("grpc-default-executor");
+  }
+
+  /**
+   * Extract the name of the channel executor thread by instantiating a channel and issuing a fake
+   * call.
+   */
+  private static String extractExecutorThreadName(InstantiatingGrpcChannelProvider channelProvider)
+      throws IOException, ExecutionException, InterruptedException {
+    GrpcTransportChannel transportChannel =
+        (GrpcTransportChannel) channelProvider.getTransportChannel();
+    try {
+      Channel channel = transportChannel.getChannel();
+
+      ClientCall<Color, Money> call =
+          channel.newCall(FakeServiceGrpc.METHOD_RECOGNIZE, CallOptions.DEFAULT);
+      Color request = Color.getDefaultInstance();
+
+      final SettableFuture<String> threadNameFuture = SettableFuture.create();
+
+      // Issue a call just to get the thread name of the channel executor
+      ClientCalls.asyncUnaryCall(
+          call,
+          request,
+          new StreamObserver<Money>() {
+            @Override
+            public void onNext(Money ignored) {
+              threadNameFuture.set(Thread.currentThread().getName());
+            }
+
+            @Override
+            public void onError(Throwable ignored) {
+              threadNameFuture.set(Thread.currentThread().getName());
+            }
+
+            @Override
+            public void onCompleted() {
+              threadNameFuture.set(Thread.currentThread().getName());
+            }
+          });
+      return threadNameFuture.get();
+    } finally {
+      transportChannel.shutdownNow();
+      transportChannel.awaitTermination(10, TimeUnit.SECONDS);
+    }
+  }
 }

gax-grpc/src/test/java/com/google/api/gax/grpc/testing/LocalChannelProvider.java

@@ -50,6 +50,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.regex.Pattern;
 
@@ -78,8 +79,15 @@ public boolean needsExecutor() {
     return false;
   }
 
+  /** @deprecated Please use {@link #withExecutor(Executor) } */
+  @Deprecated
   @Override
   public TransportChannelProvider withExecutor(ScheduledExecutorService executor) {
+    return withExecutor((Executor) executor);
+  }
+
+  @Override
+  public TransportChannelProvider withExecutor(Executor executor) {
     throw new UnsupportedOperationException("LocalChannelProvider doesn't need an executor");
   }