chore: configure gRPC session streams with DirectExecutor

CallOptions.withExecutor(MoreExecutors.directExecutor()) on the session
stream so Netty I/O threads deliver SessionStream.Listener callbacks
directly; sessionSyncContext immediately trampolines off them.

Author: igorbernstein2

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/ChannelPoolDpImpl.java

@@ -26,6 +26,7 @@
 import com.google.common.base.Ticker;
 import com.google.common.collect.HashMultiset;
 import com.google.common.collect.Multiset;
+import com.google.common.util.concurrent.MoreExecutors;
 import io.grpc.CallOptions;
 import io.grpc.ClientCall;
 import io.grpc.ManagedChannel;
@@ -246,8 +247,11 @@ public synchronized SessionStream newStream(
     channelWrapper.group.numStreams++;
     totalStreams++;
 
+    // DirectExecutor: gRPC/Netty delivers SessionStream.Listener callbacks directly on the
+    // I/O thread. All work must be fast and non-blocking; blocking work goes to sessionSyncContext.
     ClientCall<SessionRequest, SessionResponse> innerCall =
-        channelWrapper.channel.newCall(desc, callOptions);
+        channelWrapper.channel.newCall(
+            desc, callOptions.withExecutor(MoreExecutors.directExecutor()));
 
     return new SessionStreamImpl(innerCall) {
       // mark as null so that onClose can tell if onBeforeSessionStart was never called

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/SessionStream.java

@@ -37,6 +37,14 @@ public interface SessionStream {
 
   public void forceClose(@Nullable String message, @Nullable Throwable cause);
 
+  /**
+   * Callbacks for session stream events.
+   *
+   * <p><b>Invariant:</b> callbacks are delivered on Netty I/O threads via {@code DirectExecutor}.
+   * All work must be fast and non-blocking — any user-facing or potentially blocking work must be
+   * dispatched onto the session {@code SynchronizationContext} (which then forwards to the op
+   * executor) before returning. Violating this stalls the channel.
+   */
   public interface Listener {
     void onBeforeSessionStart(PeerInfo peerInfo);
 

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/SingleChannelPool.java

@@ -19,6 +19,7 @@
 import com.google.bigtable.v2.SessionClientConfiguration.ChannelPoolConfiguration;
 import com.google.bigtable.v2.SessionRequest;
 import com.google.bigtable.v2.SessionResponse;
+import com.google.common.util.concurrent.MoreExecutors;
 import io.grpc.CallOptions;
 import io.grpc.ManagedChannel;
 import io.grpc.MethodDescriptor;
@@ -45,7 +46,10 @@ public void close() {
   @Override
   public SessionStream newStream(
       MethodDescriptor<SessionRequest, SessionResponse> desc, CallOptions callOptions) {
-    return new SessionStreamImpl(channel.newCall(desc, callOptions));
+    // DirectExecutor: gRPC/Netty delivers SessionStream.Listener callbacks directly on the
+    // I/O thread. All work must be fast and non-blocking; blocking work goes to sessionSyncContext.
+    return new SessionStreamImpl(
+        channel.newCall(desc, callOptions.withExecutor(MoreExecutors.directExecutor())));
   }
 
   @Override