stub: add withExecutor API

stub/src/main/java/io/grpc/stub/AbstractStub.java

@@ -27,6 +27,7 @@ import io.grpc.ClientInterceptors;
 import io.grpc.Deadline;
 import io.grpc.ExperimentalApi;
 import io.grpc.ManagedChannelBuilder;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
@@ -120,6 +121,18 @@ public abstract class AbstractStub<S extends AbstractStub<S>> {
     return build(channel, callOptions.withDeadlineAfter(duration, unit));
   }
 
+  /**
+   * Returns a new stub with the given executor that is to be used instead of the default one
+   * specified with {@link ManagedChannelBuilder#executor}. Note that setting this option may not
+   * take effect for blocking calls.
+   *
+   * @since 1.8.0
+   */
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/3605")
+  public final S withExecutor(Executor executor) {
+    return build(channel, callOptions.withExecutor(executor));
+  }
+
   /**
    *  Set's the compressor name to use for the call.  It is the responsibility of the application
    *  to make sure the server supports decoding the compressor picked by the client.  To be clear,

stub/src/test/java/io/grpc/stub/AbstractStubTest.java

@@ -16,11 +16,15 @@
 
 package io.grpc.stub;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
 
 import io.grpc.CallOptions;
 import io.grpc.Channel;
+import java.util.concurrent.Executor;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -80,4 +84,18 @@ public class AbstractStubTest {
       return new NoopStub(channel, callOptions);
     }
   }
+
+  @Test
+  public void withExecutor() {
+    NoopStub stub = new NoopStub(channel);
+    CallOptions callOptions = stub.getCallOptions();
+
+    assertNull(callOptions.getExecutor());
+
+    Executor executor = mock(Executor.class);
+    stub = stub.withExecutor(executor);
+    callOptions = stub.getCallOptions();
+
+    assertEquals(callOptions.getExecutor(), executor);
+  }
 }