api,core: Adds an Executor field to NameResolver.Args.

Adds an Executor to NameResolver.Args, which is optionally set on ManagedChannelBuilder. This allows NameResolver implementations to avoid creating their own thread pools if the application already manages its own pools.

Addresses #3703.

api/src/main/java/io/grpc/ForwardingChannelBuilder.java

@@ -71,6 +71,12 @@ public abstract class ForwardingChannelBuilder<T extends ForwardingChannelBuilde
     return thisT();
   }
 
+  @Override
+  public T blockingExecutor(Executor executor) {
+    delegate().blockingExecutor(executor);
+    return thisT();
+  }
+
   @Override
   public T intercept(List<ClientInterceptor> interceptors) {
     delegate().intercept(interceptors);

api/src/main/java/io/grpc/ManagedChannelBuilder.java

@@ -103,6 +103,23 @@ public abstract class ManagedChannelBuilder<T extends ManagedChannelBuilder<T>>
    */
   public abstract T executor(Executor executor);
 
+  /**
+   * Provides a custom executor that will be used for operations that block.
+   *
+   * <p>It's an optional parameter. If the user has not provided an executor when the channel is
+   * built, the builder will use a static cached thread pool.
+   *
+   * <p>The channel won't take ownership of the given executor. It's caller's responsibility to shut
+   * down the executor when it's desired.
+   *
+   * @return this
+   * @throws UnsupportedOperationException if unsupported
+   * @since 1.25.0
+   */
+  public T blockingExecutor(Executor executor) {
+    throw new UnsupportedOperationException();
+  }
+
   /**
    * Adds interceptors that will be called before the channel performs its real work. This is
    * functionally equivalent to using {@link ClientInterceptors#intercept(Channel, List)}, but while

api/src/main/java/io/grpc/NameResolver.java

@@ -29,6 +29,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Executor;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 
@@ -411,13 +412,19 @@ public abstract class NameResolver {
     private final ProxyDetector proxyDetector;
     private final SynchronizationContext syncContext;
     private final ServiceConfigParser serviceConfigParser;
+    @Nullable private final Executor executor;
 
-    Args(Integer defaultPort, ProxyDetector proxyDetector,
-        SynchronizationContext syncContext, ServiceConfigParser serviceConfigParser) {
+    Args(
+        Integer defaultPort,
+        ProxyDetector proxyDetector,
+        SynchronizationContext syncContext,
+        ServiceConfigParser serviceConfigParser,
+        @Nullable Executor executor) {
       this.defaultPort = checkNotNull(defaultPort, "defaultPort not set");
       this.proxyDetector = checkNotNull(proxyDetector, "proxyDetector not set");
       this.syncContext = checkNotNull(syncContext, "syncContext not set");
       this.serviceConfigParser = checkNotNull(serviceConfigParser, "serviceConfigParser not set");
+      this.executor = executor;
     }
 
     /**
@@ -459,6 +466,17 @@ public abstract class NameResolver {
       return serviceConfigParser;
     }
 
+    /**
+     * Returns the Executor on which this resolver should execute long-running or I/O bound work.
+     * Null if no Executor was set.
+     *
+     * @since 1.25.0
+     */
+    @Nullable
+    public Executor getBlockingExecutor() {
+      return executor;
+    }
+
     @Override
     public String toString() {
       return MoreObjects.toStringHelper(this)
@@ -466,6 +484,7 @@ public abstract class NameResolver {
           .add("proxyDetector", proxyDetector)
           .add("syncContext", syncContext)
           .add("serviceConfigParser", serviceConfigParser)
+          .add("executor", executor)
           .toString();
     }
 
@@ -480,6 +499,7 @@ public abstract class NameResolver {
       builder.setProxyDetector(proxyDetector);
       builder.setSynchronizationContext(syncContext);
       builder.setServiceConfigParser(serviceConfigParser);
+      builder.setBlockingExecutor(executor);
       return builder;
     }
 
@@ -502,6 +522,7 @@ public abstract class NameResolver {
       private ProxyDetector proxyDetector;
       private SynchronizationContext syncContext;
       private ServiceConfigParser serviceConfigParser;
+      private Executor executor;
 
       Builder() {
       }
@@ -546,13 +567,23 @@ public abstract class NameResolver {
         return this;
       }
 
+      /**
+       * See {@link Args#getBlockingExecutor}. This is an optional field.
+       *
+       * @since 1.25.0
+       */
+      public Builder setBlockingExecutor(Executor executor) {
+        this.executor = executor;
+        return this;
+      }
+
       /**
        * Builds an {@link Args}.
        *
        * @since 1.21.0
        */
       public Args build() {
-        return new Args(defaultPort, proxyDetector, syncContext, serviceConfigParser);
+        return new Args(defaultPort, proxyDetector, syncContext, serviceConfigParser, executor);
       }
     }
   }

api/src/test/java/io/grpc/NameResolverTest.java

@@ -29,6 +29,8 @@ import java.lang.Thread.UncaughtExceptionHandler;
 import java.net.URI;
 import java.util.Collections;
 import java.util.Map;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicReference;
 import org.junit.Before;
 import org.junit.Test;
@@ -43,6 +45,7 @@ public class NameResolverTest {
   private final SynchronizationContext syncContext =
       new SynchronizationContext(mock(UncaughtExceptionHandler.class));
   private final ServiceConfigParser parser = mock(ServiceConfigParser.class);
+  private final Executor executor = Executors.newSingleThreadExecutor();
   private URI uri;
   private final NameResolver nameResolver = mock(NameResolver.class);
 
@@ -58,12 +61,14 @@ public class NameResolverTest {
     assertThat(args.getProxyDetector()).isSameInstanceAs(proxyDetector);
     assertThat(args.getSynchronizationContext()).isSameInstanceAs(syncContext);
     assertThat(args.getServiceConfigParser()).isSameInstanceAs(parser);
+    assertThat(args.getBlockingExecutor()).isSameInstanceAs(executor);
 
     NameResolver.Args args2 = args.toBuilder().build();
     assertThat(args2.getDefaultPort()).isEqualTo(defaultPort);
     assertThat(args2.getProxyDetector()).isSameInstanceAs(proxyDetector);
     assertThat(args2.getSynchronizationContext()).isSameInstanceAs(syncContext);
     assertThat(args2.getServiceConfigParser()).isSameInstanceAs(parser);
+    assertThat(args2.getBlockingExecutor()).isSameInstanceAs(executor);
 
     assertThat(args2).isNotSameInstanceAs(args);
     assertThat(args2).isNotEqualTo(args);
@@ -246,6 +251,7 @@ public class NameResolverTest {
         .setProxyDetector(proxyDetector)
         .setSynchronizationContext(syncContext)
         .setServiceConfigParser(parser)
+        .setBlockingExecutor(executor)
         .build();
   }
 }

core/src/main/java/io/grpc/internal/AbstractManagedChannelImplBuilder.java

@@ -95,6 +95,8 @@ public abstract class AbstractManagedChannelImplBuilder
 
   ObjectPool<? extends Executor> executorPool = DEFAULT_EXECUTOR_POOL;
 
+  ObjectPool<? extends Executor> blockingExecutorPool = DEFAULT_EXECUTOR_POOL;
+
   private final List<ClientInterceptor> interceptors = new ArrayList<>();
   final NameResolverRegistry nameResolverRegistry = NameResolverRegistry.getDefaultRegistry();
 
@@ -217,6 +219,16 @@ public abstract class AbstractManagedChannelImplBuilder
     return thisT();
   }
 
+  @Override
+  public final T blockingExecutor(Executor executor) {
+    if (executor != null) {
+      this.blockingExecutorPool = new FixedObjectPool<>(executor);
+    } else {
+      this.blockingExecutorPool = DEFAULT_EXECUTOR_POOL;
+    }
+    return thisT();
+  }
+
   @Override
   public final T intercept(List<ClientInterceptor> interceptors) {
     this.interceptors.addAll(interceptors);

core/src/main/java/io/grpc/internal/DnsNameResolver.java

@@ -138,6 +138,8 @@ final class DnsNameResolver extends NameResolver {
   private final String authority;
   private final String host;
   private final int port;
+
+  /** Executor that will be used if an Executor is not provide via {@link NameResolver.Args}. */
   private final Resource<Executor> executorResource;
   private final long cacheTtlNanos;
   private final SynchronizationContext syncContext;
@@ -147,6 +149,10 @@ final class DnsNameResolver extends NameResolver {
   private ResolutionResults cachedResolutionResults;
   private boolean shutdown;
   private Executor executor;
+
+  /** True if using an executor resource that should be released after use. */
+  private final boolean usingExecutorResource;
+
   private boolean resolving;
 
   // The field must be accessed from syncContext, although the methods on an Listener2 can be called
@@ -176,6 +182,8 @@ final class DnsNameResolver extends NameResolver {
     this.stopwatch = Preconditions.checkNotNull(stopwatch, "stopwatch");
     this.syncContext =
         Preconditions.checkNotNull(args.getSynchronizationContext(), "syncContext");
+    this.executor = args.getBlockingExecutor();
+    this.usingExecutorResource = executor == null;
   }
 
   @Override
@@ -186,7 +194,9 @@ final class DnsNameResolver extends NameResolver {
   @Override
   public void start(Listener2 listener) {
     Preconditions.checkState(this.listener == null, "already started");
-    executor = SharedResourceHolder.get(executorResource);
+    if (usingExecutorResource) {
+      executor = SharedResourceHolder.get(executorResource);
+    }
     this.listener = Preconditions.checkNotNull(listener, "listener");
     resolve();
   }
@@ -361,7 +371,7 @@ final class DnsNameResolver extends NameResolver {
       return;
     }
     shutdown = true;
-    if (executor != null) {
+    if (executor != null && usingExecutorResource) {
       executor = SharedResourceHolder.release(executorResource, executor);
     }
   }

core/src/main/java/io/grpc/internal/ManagedChannelImpl.java

@@ -142,6 +142,7 @@ final class ManagedChannelImpl extends ManagedChannel implements
   private final ObjectPool<? extends Executor> executorPool;
   private final ObjectPool<? extends Executor> balancerRpcExecutorPool;
   private final ExecutorHolder balancerRpcExecutorHolder;
+  private final ExecutorHolder blockingExecutorHolder;
   private final TimeProvider timeProvider;
   private final int maxTraceEvents;
 
@@ -565,16 +566,30 @@ final class ManagedChannelImpl extends ManagedChannel implements
         builder.proxyDetector != null ? builder.proxyDetector : GrpcUtil.DEFAULT_PROXY_DETECTOR;
     this.retryEnabled = builder.retryEnabled && !builder.temporarilyDisableRetry;
     this.loadBalancerFactory = new AutoConfiguredLoadBalancerFactory(builder.defaultLbPolicy);
+    this.blockingExecutorHolder =
+        new ExecutorHolder(
+            checkNotNull(builder.blockingExecutorPool, "blockingExecutorPool"));
     this.nameResolverRegistry = builder.nameResolverRegistry;
-    this.nameResolverArgs = NameResolver.Args.newBuilder()
-        .setDefaultPort(builder.getDefaultPort())
-        .setProxyDetector(proxyDetector)
-        .setSynchronizationContext(syncContext)
-        .setServiceConfigParser(
-            new ScParser(
-                retryEnabled, builder.maxRetryAttempts, builder.maxHedgedAttempts,
-                loadBalancerFactory))
-        .build();
+    this.nameResolverArgs =
+        NameResolver.Args.newBuilder()
+            .setDefaultPort(builder.getDefaultPort())
+            .setProxyDetector(proxyDetector)
+            .setSynchronizationContext(syncContext)
+            .setServiceConfigParser(
+                new ScParser(
+                    retryEnabled,
+                    builder.maxRetryAttempts,
+                    builder.maxHedgedAttempts,
+                    loadBalancerFactory))
+            .setBlockingExecutor(
+                // Avoid creating the blockingExecutor until it is first used
+                new Executor() {
+                  @Override
+                  public void execute(Runnable command) {
+                    blockingExecutorHolder.getExecutor().execute(command);
+                  }
+                })
+            .build();
     this.nameResolver = getNameResolver(target, nameResolverFactory, nameResolverArgs);
     this.timeProvider = checkNotNull(timeProvider, "timeProvider");
     maxTraceEvents = builder.maxTraceEvents;
@@ -885,6 +900,7 @@ final class ManagedChannelImpl extends ManagedChannel implements
       terminatedLatch.countDown();
       executorPool.returnObject(executor);
       balancerRpcExecutorHolder.release();
+      blockingExecutorHolder.release();
       // Release the transport factory so that it can deallocate any resources.
       transportFactory.close();
     }

core/src/test/java/io/grpc/internal/AbstractManagedChannelImplBuilderTest.java

@@ -99,6 +99,21 @@ public class AbstractManagedChannelImplBuilderTest {
     assertEquals(MoreExecutors.directExecutor(), builder.executorPool.getObject());
   }
 
+  @Test
+  public void blockingExecutor_normal() {
+    Executor executor = mock(Executor.class);
+    assertEquals(builder, builder.blockingExecutor(executor));
+    assertEquals(executor, builder.blockingExecutorPool.getObject());
+  }
+
+  @Test
+  public void blockingExecutor_null() {
+    ObjectPool<? extends Executor> defaultValue = builder.blockingExecutorPool;
+    builder.blockingExecutor(mock(Executor.class));
+    assertEquals(builder, builder.blockingExecutor(null));
+    assertEquals(defaultValue, builder.blockingExecutorPool);
+  }
+
   @Test
   public void nameResolverFactory_default() {
     assertNotNull(builder.getNameResolverFactory());

core/src/test/java/io/grpc/internal/DnsNameResolverTest.java

@@ -70,6 +70,7 @@ import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.regex.Pattern;
@@ -120,17 +121,20 @@ public class DnsNameResolverTest {
   private final FakeClock fakeClock = new FakeClock();
   private final FakeClock fakeExecutor = new FakeClock();
 
-  private final Resource<Executor> fakeExecutorResource =
-      new Resource<Executor>() {
-        @Override
-        public Executor create() {
-          return fakeExecutor.getScheduledExecutorService();
-        }
+  private final FakeExecutorResource fakeExecutorResource = new FakeExecutorResource();
 
-        @Override
-        public void close(Executor instance) {
-        }
-      };
+  private final class FakeExecutorResource implements Resource<Executor> {
+    private final AtomicInteger createCount = new AtomicInteger();
+
+    @Override
+    public Executor create() {
+      createCount.incrementAndGet();
+      return fakeExecutor.getScheduledExecutorService();
+    }
+
+    @Override
+    public void close(Executor instance) {}
+  }
 
   @Mock
   private NameResolver.Listener2 mockListener;
@@ -165,18 +169,20 @@ public class DnsNameResolverTest {
       final ProxyDetector proxyDetector,
       Stopwatch stopwatch,
       boolean isAndroid) {
-    DnsNameResolver dnsResolver = new DnsNameResolver(
-        null,
-        name,
+    NameResolver.Args args =
         NameResolver.Args.newBuilder()
             .setDefaultPort(defaultPort)
             .setProxyDetector(proxyDetector)
             .setSynchronizationContext(syncContext)
             .setServiceConfigParser(mock(ServiceConfigParser.class))
-            .build(),
-        fakeExecutorResource,
-        stopwatch,
-        isAndroid);
+            .build();
+    return newResolver(name, stopwatch, isAndroid, args);
+  }
+
+  private DnsNameResolver newResolver(
+      String name, Stopwatch stopwatch, boolean isAndroid, NameResolver.Args args) {
+    DnsNameResolver dnsResolver =
+        new DnsNameResolver(null, name, args, fakeExecutorResource, stopwatch, isAndroid);
     // By default, using the mocked ResourceResolver to avoid I/O
     dnsResolver.setResourceResolver(new JndiResourceResolver(recordFetcher));
     return dnsResolver;
@@ -293,6 +299,65 @@ public class DnsNameResolverTest {
     verify(mockResolver, times(2)).resolveAddress(anyString());
   }
 
+  @Test
+  public void testExecutor_default() throws Exception {
+    final List<InetAddress> answer = createAddressList(2);
+
+    DnsNameResolver resolver = newResolver("foo.googleapis.com", 81);
+    AddressResolver mockResolver = mock(AddressResolver.class);
+    when(mockResolver.resolveAddress(anyString())).thenReturn(answer);
+    resolver.setAddressResolver(mockResolver);
+
+    resolver.start(mockListener);
+    assertEquals(1, fakeExecutor.runDueTasks());
+    verify(mockListener).onResult(resultCaptor.capture());
+    assertAnswerMatches(answer, 81, resultCaptor.getValue());
+    assertEquals(0, fakeClock.numPendingTasks());
+
+    resolver.shutdown();
+
+    assertThat(fakeExecutorResource.createCount.get()).isEqualTo(1);
+  }
+
+  @Test
+  public void testExecutor_custom() throws Exception {
+    final List<InetAddress> answer = createAddressList(2);
+    final AtomicInteger executions = new AtomicInteger();
+
+    NameResolver.Args args =
+        NameResolver.Args.newBuilder()
+            .setDefaultPort(81)
+            .setProxyDetector(GrpcUtil.NOOP_PROXY_DETECTOR)
+            .setSynchronizationContext(syncContext)
+            .setServiceConfigParser(mock(ServiceConfigParser.class))
+            .setBlockingExecutor(
+                new Executor() {
+                  @Override
+                  public void execute(Runnable command) {
+                    executions.incrementAndGet();
+                    command.run();
+                  }
+                })
+            .build();
+
+    DnsNameResolver resolver =
+        newResolver("foo.googleapis.com", Stopwatch.createUnstarted(), false, args);
+    AddressResolver mockResolver = mock(AddressResolver.class);
+    when(mockResolver.resolveAddress(anyString())).thenReturn(answer);
+    resolver.setAddressResolver(mockResolver);
+
+    resolver.start(mockListener);
+    assertEquals(0, fakeExecutor.runDueTasks());
+    verify(mockListener).onResult(resultCaptor.capture());
+    assertAnswerMatches(answer, 81, resultCaptor.getValue());
+    assertEquals(0, fakeClock.numPendingTasks());
+
+    resolver.shutdown();
+
+    assertThat(fakeExecutorResource.createCount.get()).isEqualTo(0);
+    assertThat(executions.get()).isEqualTo(1);
+  }
+
   @Test
   public void resolveAll_failsOnEmptyResult() {
     DnsNameResolver nr = newResolver("dns:///addr.fake:1234", 443);

core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java

@@ -265,6 +265,8 @@ public class ManagedChannelImplTest {
   private ObjectPool<Executor> balancerRpcExecutorPool;
   @Mock
   private CallCredentials creds;
+  @Mock
+  private Executor blockingExecutor;
   private ChannelBuilder channelBuilder;
   private boolean requestConnection = true;
   private BlockingQueue<MockClientTransportInfo> transports;
@@ -319,11 +321,14 @@ public class ManagedChannelImplTest {
     when(balancerRpcExecutorPool.getObject())
         .thenReturn(balancerRpcExecutor.getScheduledExecutorService());
 
-    channelBuilder = new ChannelBuilder()
-        .nameResolverFactory(new FakeNameResolverFactory.Builder(expectedUri).build())
-        .defaultLoadBalancingPolicy(MOCK_POLICY_NAME)
-        .userAgent(USER_AGENT)
-        .idleTimeout(AbstractManagedChannelImplBuilder.IDLE_MODE_MAX_TIMEOUT_DAYS, TimeUnit.DAYS);
+    channelBuilder =
+        new ChannelBuilder()
+            .nameResolverFactory(new FakeNameResolverFactory.Builder(expectedUri).build())
+            .defaultLoadBalancingPolicy(MOCK_POLICY_NAME)
+            .userAgent(USER_AGENT)
+            .idleTimeout(
+                AbstractManagedChannelImplBuilder.IDLE_MODE_MAX_TIMEOUT_DAYS, TimeUnit.DAYS)
+            .blockingExecutor(blockingExecutor);
     channelBuilder.executorPool = executorPool;
     channelBuilder.binlog = null;
     channelBuilder.channelz = channelz;
@@ -3582,6 +3587,15 @@ public class ManagedChannelImplTest {
     assertThat(args).isNotNull();
     assertThat(args.getDefaultPort()).isEqualTo(DEFAULT_PORT);
     assertThat(args.getProxyDetector()).isSameInstanceAs(neverProxy);
+
+    verify(blockingExecutor, never()).execute(any(Runnable.class));
+    args.getBlockingExecutor()
+        .execute(
+            new Runnable() {
+              @Override
+              public void run() {}
+            });
+    verify(blockingExecutor, times(1)).execute(any(Runnable.class));
   }
 
   @Test