core: ManagedChannelImpl2.

core/src/main/java/io/grpc/LoadBalancer2.java

@@ -34,6 +34,7 @@
 import com.google.common.base.Preconditions;
 
 import java.util.List;
+import java.util.concurrent.Executor;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.Immutable;
 import javax.annotation.concurrent.NotThreadSafe;
@@ -337,9 +338,13 @@ public abstract static class Helper {
      *
      * <p>The LoadBalancer is responsible for closing unused OOB channels, and closing all OOB
      * channels within {@link #shutdown}.
+     *
+     * @param eag the address(es) of the channel
+     * @param authority the authority of the destination this channel connects to
+     * @param executor the default executor for running RPC callbacks
      */
     public abstract ManagedChannel createOobChannel(
-        EquivalentAddressGroup eag, String authority);
+        EquivalentAddressGroup eag, String authority, Executor executor);
 
     /**
      * Set a new picker to the channel.

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

@@ -31,6 +31,8 @@
 
 package io.grpc.internal;
 
+import static com.google.common.base.Preconditions.checkNotNull;
+
 import com.google.common.annotations.VisibleForTesting;
 
 import java.util.LinkedList;
@@ -97,7 +99,7 @@ void drain() {
    */
   ChannelExecutor executeLater(Runnable runnable) {
     synchronized (lock) {
-      queue.add(runnable);
+      queue.add(checkNotNull(runnable, "runnable is null"));
     }
     return this;
   }

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

@@ -127,7 +127,7 @@ interface ClientTransportProvider {
     /**
      * Returns a transport for a new call.
      */
-    ClientTransport get(CallOptions callOptions);
+    ClientTransport get(CallOptions callOptions, Metadata headers);
   }
 
   ClientCallImpl<ReqT, RespT> setDecompressorRegistry(DecompressorRegistry decompressorRegistry) {
@@ -215,7 +215,7 @@ public void runInContext() {
     if (!deadlineExceeded) {
       updateTimeoutHeaders(effectiveDeadline, callOptions.getDeadline(),
           context.getDeadline(), headers);
-      ClientTransport transport = clientTransportProvider.get(callOptions);
+      ClientTransport transport = clientTransportProvider.get(callOptions, headers);
       Context origContext = context.attach();
       try {
         stream = transport.newStream(method, headers, callOptions, statsTraceCtx);

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

@@ -51,6 +51,7 @@
 import io.grpc.EquivalentAddressGroup;
 import io.grpc.LoadBalancer;
 import io.grpc.ManagedChannel;
+import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.NameResolver;
 import io.grpc.ResolvedServerInfoGroup;
@@ -359,7 +360,7 @@ private void rescheduleIdleTimer() {
 
   private final ClientTransportProvider transportProvider = new ClientTransportProvider() {
     @Override
-    public ClientTransport get(CallOptions callOptions) {
+    public ClientTransport get(CallOptions callOptions, Metadata headers) {
       LoadBalancer<ClientTransport> balancer = loadBalancer;
       if (balancer == null) {
         // Current state is either idle or in grace period