core: Use offloadExecutor for CallCredentials (#9263)

Change the construction of CallCredentialsApplyingTransportFactory in
ManagedChannelImpl to use the offloadExecutor as indicated in
https://github.com/grpc/grpc-java/issues/6279#issuecomment-1147507365 .

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

@@ -621,10 +621,12 @@ final class ManagedChannelImpl extends ManagedChannel implements
     this.executor = checkNotNull(executorPool.getObject(), "executor");
     this.originalChannelCreds = builder.channelCredentials;
     this.originalTransportFactory = clientTransportFactory;
+    this.offloadExecutorHolder =
+        new ExecutorHolder(checkNotNull(builder.offloadExecutorPool, "offloadExecutorPool"));
     this.transportFactory = new CallCredentialsApplyingTransportFactory(
-        clientTransportFactory, builder.callCredentials, this.executor);
+        clientTransportFactory, builder.callCredentials, this.offloadExecutorHolder);
     this.oobTransportFactory = new CallCredentialsApplyingTransportFactory(
-        clientTransportFactory, null, this.executor);
+        clientTransportFactory, null, this.offloadExecutorHolder);
     this.scheduledExecutor =
         new RestrictedScheduledExecutor(transportFactory.getScheduledExecutorService());
     maxTraceEvents = builder.maxTraceEvents;
@@ -636,9 +638,6 @@ final class ManagedChannelImpl extends ManagedChannel implements
         builder.proxyDetector != null ? builder.proxyDetector : GrpcUtil.DEFAULT_PROXY_DETECTOR;
     this.retryEnabled = builder.retryEnabled;
     this.loadBalancerFactory = new AutoConfiguredLoadBalancerFactory(builder.defaultLbPolicy);
-    this.offloadExecutorHolder =
-        new ExecutorHolder(
-            checkNotNull(builder.offloadExecutorPool, "offloadExecutorPool"));
     this.nameResolverRegistry = builder.nameResolverRegistry;
     ScParser serviceConfigParser =
         new ScParser(
@@ -654,14 +653,7 @@ final class ManagedChannelImpl extends ManagedChannel implements
             .setScheduledExecutorService(scheduledExecutor)
             .setServiceConfigParser(serviceConfigParser)
             .setChannelLogger(channelLogger)
-            .setOffloadExecutor(
+            .setOffloadExecutor(this.offloadExecutorHolder)
-                // Avoid creating the offloadExecutor until it is first used
-                new Executor() {
-                  @Override
-                  public void execute(Runnable command) {
-                    offloadExecutorHolder.getExecutor().execute(command);
-                  }
-                })
             .build();
     this.authorityOverride = builder.authorityOverride;
     this.nameResolverFactory = builder.nameResolverFactory;
@@ -2219,8 +2211,10 @@ final class ManagedChannelImpl extends ManagedChannel implements
 
   /**
    * Lazily request for Executor from an executor pool.
+   * Also act as an Executor directly to simply run a cmd
    */
-  private static final class ExecutorHolder {
+  @VisibleForTesting
+  static final class ExecutorHolder implements Executor {
     private final ObjectPool<? extends Executor> pool;
     private Executor executor;
 
@@ -2240,6 +2234,11 @@ final class ManagedChannelImpl extends ManagedChannel implements
         executor = pool.returnObject(executor);
       }
     }
+
+    @Override
+    public void execute(Runnable command) {
+      getExecutor().execute(command);
+    }
   }
 
   private static final class RestrictedScheduledExecutor implements ScheduledExecutorService {

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

@@ -2396,9 +2396,12 @@ public class ManagedChannelImplTest {
     updateBalancingStateSafely(helper, READY, mockPicker);
     executor.runDueTasks();
     ArgumentCaptor<RequestInfo> infoCaptor = ArgumentCaptor.forClass(null);
+    ArgumentCaptor<Executor> executorArgumentCaptor = ArgumentCaptor.forClass(null);
     ArgumentCaptor<CallCredentials.MetadataApplier> applierCaptor = ArgumentCaptor.forClass(null);
     verify(creds).applyRequestMetadata(infoCaptor.capture(),
-        same(executor.getScheduledExecutorService()), applierCaptor.capture());
+        executorArgumentCaptor.capture(), applierCaptor.capture());
+    assertSame(offloadExecutor,
+            ((ManagedChannelImpl.ExecutorHolder) executorArgumentCaptor.getValue()).getExecutor());
     assertEquals("testValue", testKey.get(credsApplyContexts.poll()));
     assertEquals(AUTHORITY, infoCaptor.getValue().getAuthority());
     assertEquals(SecurityLevel.NONE, infoCaptor.getValue().getSecurityLevel());
@@ -2423,7 +2426,9 @@ public class ManagedChannelImplTest {
     call.start(mockCallListener, new Metadata());
 
     verify(creds, times(2)).applyRequestMetadata(infoCaptor.capture(),
-        same(executor.getScheduledExecutorService()), applierCaptor.capture());
+            executorArgumentCaptor.capture(), applierCaptor.capture());
+    assertSame(offloadExecutor,
+            ((ManagedChannelImpl.ExecutorHolder) executorArgumentCaptor.getValue()).getExecutor());
     assertEquals("testValue", testKey.get(credsApplyContexts.poll()));
     assertEquals(AUTHORITY, infoCaptor.getValue().getAuthority());
     assertEquals(SecurityLevel.NONE, infoCaptor.getValue().getSecurityLevel());