Call executor safeguard for interceptor

Author: kannanjgithub

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

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2026 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.internal;
+
+import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
+
+import java.util.concurrent.Executor;
+
+/**
+ * Common utilities for GRPC call executors.
+ */
+final class CallExecutors {
+
+  private CallExecutors() {}
+
+  /**
+   * Wraps an executor with safeguarding (serialization) if not already safeguarded.
+   */
+  static Executor safeguard(Executor executor) {
+    if (executor instanceof SerializingExecutor
+        || executor instanceof SerializeReentrantCallsDirectExecutor) {
+      return executor;
+    }
+    if (executor == directExecutor()) {
+      return new SerializeReentrantCallsDirectExecutor();
+    }
+    return new SerializingExecutor(executor);
+  }
+
+  /**
+   * Returns true if the executor is safeguarded (e.g. a {@link SerializingExecutor} or
+   * {@link SerializeReentrantCallsDirectExecutor}).
+   */
+  static boolean isSafeguarded(Executor executor) {
+    return executor instanceof SerializingExecutor
+        || executor instanceof SerializeReentrantCallsDirectExecutor;
+  }
+}

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

@@ -107,13 +107,8 @@ final class ClientCallImpl<ReqT, RespT> extends ClientCall<ReqT, RespT> {
     // If we know that the executor is a direct executor, we don't need to wrap it with a
     // SerializingExecutor. This is purely for performance reasons.
     // See https://github.com/grpc/grpc-java/issues/368
-    if (executor == directExecutor()) {
-      this.callExecutor = new SerializeReentrantCallsDirectExecutor();
-      callExecutorIsDirect = true;
-    } else {
-      this.callExecutor = new SerializingExecutor(executor);
-      callExecutorIsDirect = false;
-    }
+    this.callExecutor = CallExecutors.safeguard(executor);
+    callExecutorIsDirect = (this.callExecutor instanceof SerializeReentrantCallsDirectExecutor);
     this.channelCallsTracer = channelCallsTracer;
     // Propagate the context from the thread which initiated the call to all callbacks.
     this.context = Context.current();

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

@@ -808,6 +808,13 @@ public boolean isTerminated() {
   @Override
   public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(MethodDescriptor<ReqT, RespT> method,
       CallOptions callOptions) {
+    Executor executor = callOptions.getExecutor();
+    if (executor == null) {
+      executor = this.executor;
+    }
+    // All calls on the channel should have a safeguarded executor in CallOptions before
+    // calling interceptors.
+    callOptions = callOptions.withExecutor(CallExecutors.safeguard(executor));
     return interceptorChannel.newCall(method, callOptions);
   }
 
@@ -821,7 +828,7 @@ private Executor getCallExecutor(CallOptions callOptions) {
     if (executor == null) {
       executor = this.executor;
     }
-    return executor;
+    return CallExecutors.safeguard(executor);
   }
 
   private class RealChannel extends Channel {
@@ -1084,9 +1091,12 @@ static final class ConfigSelectingClientCall<ReqT, RespT>
       this.configSelector = configSelector;
       this.channel = channel;
       this.method = method;
-      this.callExecutor =
-          callOptions.getExecutor() == null ? channelExecutor : callOptions.getExecutor();
-      this.callOptions = callOptions.withExecutor(callExecutor);
+      Executor executor = callOptions.getExecutor();
+      if (executor == null) {
+        executor = channelExecutor;
+      }
+      this.callExecutor = CallExecutors.safeguard(executor);
+      this.callOptions = callOptions.withExecutor(this.callExecutor);
       this.context = Context.current();
     }
 

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

@@ -85,8 +85,11 @@ public ClientStream newStream(MethodDescriptor<?, ?> method,
   @Override
   public <RequestT, ResponseT> ClientCall<RequestT, ResponseT> newCall(
       MethodDescriptor<RequestT, ResponseT> methodDescriptor, CallOptions callOptions) {
-    final Executor effectiveExecutor =
-        callOptions.getExecutor() == null ? executor : callOptions.getExecutor();
+    Executor callExecutor = callOptions.getExecutor();
+    if (callExecutor == null) {
+      callExecutor = this.executor;
+    }
+    final Executor effectiveExecutor = CallExecutors.safeguard(callExecutor);
     if (callOptions.isWaitForReady()) {
       return new ClientCall<RequestT, ResponseT>() {
         @Override