Threading refactor (3/7): Executor isolation

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/AuthorizedViewAsync.java

@@ -33,6 +33,7 @@
 import io.grpc.Deadline;
 import java.io.Closeable;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 
 public class AuthorizedViewAsync implements AutoCloseable, Closeable {
 
@@ -48,7 +49,8 @@ static AuthorizedViewAsync createAndStart(
       String viewId,
       Permission permission,
       Metrics metrics,
-      BigtableTimer timer) {
+      BigtableTimer timer,
+      Executor userCallbackExecutor) {
 
     AuthorizedViewName viewName =
         AuthorizedViewName.builder()
@@ -78,7 +80,8 @@ static AuthorizedViewAsync createAndStart(
             callOptions,
             viewName.toString(),
             metrics,
-            timer);
+            timer,
+            userCallbackExecutor);
 
     return new AuthorizedViewAsync(base);
   }

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/Client.java

@@ -37,6 +37,7 @@
 import com.google.cloud.bigtable.data.v2.internal.session.NettyWheelTimer;
 import com.google.cloud.bigtable.data.v2.internal.session.SessionPool;
 import com.google.cloud.bigtable.data.v2.internal.util.ClientConfigurationManager;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import io.grpc.CallOptions;
 import io.opencensus.stats.Stats;
 import io.opencensus.tags.Tags;
@@ -45,6 +46,7 @@
 import java.util.Collections;
 import java.util.Set;
 import java.util.WeakHashMap;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -73,6 +75,11 @@ public class Client implements AutoCloseable {
   private final FeatureFlags featureFlags;
   private final ClientInfo clientInfo;
   private final Resource<ScheduledExecutorService> backgroundExecutor;
+  // Drains the per-op SerializingExecutor. Cached pool so a blocked user callback does not starve
+  // heartbeats, retry delays, or other vRPCs (which all run on backgroundExecutor).
+  // TODO: source from the gax TransportChannelProvider so transport and user-callback dispatch
+  // share the same pool. Blocked on missing APIs to extract the configured executor from gax.
+  private final Resource<ExecutorService> userCallbackExecutor;
   // Hashed-wheel timer for heartbeat / deadline / watchdog / retry scheduling. Built over
   // backgroundExecutor (the timer's tick thread dispatches bodies onto it). Single tick thread per
   // Client, shared across every SessionPoolImpl.
@@ -96,6 +103,12 @@ public static Client create(ClientSettings settings) throws IOException {
             .build();
 
     ScheduledExecutorService backgroundExecutor = Executors.newScheduledThreadPool(4);
+    ExecutorService userCallbackExecutor =
+        Executors.newCachedThreadPool(
+            new ThreadFactoryBuilder()
+                .setNameFormat("bigtable-callback-%d")
+                .setDaemon(true)
+                .build());
 
     // TODO: compat layer: get this from settings
     String universeDomain = "googleapis.com";
@@ -143,6 +156,7 @@ public static Client create(ClientSettings settings) throws IOException {
       }
       metrics.close();
       backgroundExecutor.shutdown();
+      userCallbackExecutor.shutdown();
       throw new RuntimeException("Failed to fetch initial config", e);
     }
 
@@ -156,7 +170,8 @@ public static Client create(ClientSettings settings) throws IOException {
         settings.getChannelProvider(),
         Resource.createOwned(metrics, metrics::close),
         Resource.createOwned(configManager, configManager::close),
-        Resource.createOwned(backgroundExecutor, backgroundExecutor::shutdown));
+        Resource.createOwned(backgroundExecutor, backgroundExecutor::shutdown),
+        Resource.createOwned(userCallbackExecutor, userCallbackExecutor::shutdown));
   }
 
   public Client(
@@ -165,13 +180,15 @@ public Client(
       ChannelProvider channelProvider,
       Resource<Metrics> metrics,
       Resource<ClientConfigurationManager> configManager,
-      Resource<ScheduledExecutorService> bgExecutor)
+      Resource<ScheduledExecutorService> bgExecutor,
+      Resource<ExecutorService> userCallbackExecutor)
       throws IOException {
     this.featureFlags = featureFlags;
     this.clientInfo = clientInfo;
     this.metrics = metrics;
     this.configManager = configManager;
     this.backgroundExecutor = bgExecutor;
+    this.userCallbackExecutor = userCallbackExecutor;
     // Timer's tick thread dispatches bodies onto backgroundExecutor — tick-thread-blocking work
     // (anything that takes a pool lock) gets handed off there instead of stalling the wheel.
     this.sessionTimer = new NettyWheelTimer("bigtable-session-timer", bgExecutor.get());
@@ -214,6 +231,7 @@ public void close() {
     // Stop the timer before tearing down backgroundExecutor (the timer's dispatcher).
     sessionTimer.stop();
     backgroundExecutor.close();
+    userCallbackExecutor.close();
   }
 
   public TableAsync openTableAsync(String tableId, Permission permission) {
@@ -227,7 +245,8 @@ public TableAsync openTableAsync(String tableId, Permission permission) {
             tableId,
             permission,
             metrics.get(),
-            sessionTimer);
+            sessionTimer,
+            userCallbackExecutor.get());
     sessionPools.add(tableAsync.getSessionPool());
     return tableAsync;
   }
@@ -245,7 +264,8 @@ public AuthorizedViewAsync openAuthorizedViewAsync(
             viewId,
             permission,
             metrics.get(),
-            sessionTimer);
+            sessionTimer,
+            userCallbackExecutor.get());
     sessionPools.add(viewAsync.getSessionPool());
     return viewAsync;
   }
@@ -262,7 +282,8 @@ public MaterializedViewAsync openMaterializedViewAsync(
             viewId,
             permission,
             metrics.get(),
-            sessionTimer);
+            sessionTimer,
+            userCallbackExecutor.get());
     sessionPools.add(viewAsync.getSessionPool());
     return viewAsync;
   }

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/MaterializedViewAsync.java

@@ -30,6 +30,7 @@
 import io.grpc.Deadline;
 import java.io.Closeable;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 
 public class MaterializedViewAsync implements AutoCloseable, Closeable {
 
@@ -44,7 +45,8 @@ public static MaterializedViewAsync createAndStart(
       String viewId,
       OpenMaterializedViewRequest.Permission permission,
       Metrics metrics,
-      BigtableTimer timer) {
+      BigtableTimer timer,
+      Executor userCallbackExecutor) {
 
     MaterializedViewName viewName =
         MaterializedViewName.builder()
@@ -73,7 +75,8 @@ public static MaterializedViewAsync createAndStart(
             callOptions,
             viewId,
             metrics,
-            timer);
+            timer,
+            userCallbackExecutor);
 
     return new MaterializedViewAsync(base);
   }

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableAsync.java

@@ -34,6 +34,7 @@
 import io.grpc.Deadline;
 import java.io.Closeable;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 
 public class TableAsync implements AutoCloseable, Closeable {
   private final TableBase base;
@@ -47,7 +48,8 @@ public static TableAsync createAndStart(
       String tableId,
       Permission permission,
       Metrics metrics,
-      BigtableTimer timer) {
+      BigtableTimer timer,
+      Executor userCallbackExecutor) {
 
     TableName tableName =
         TableName.builder()
@@ -76,7 +78,8 @@ public static TableAsync createAndStart(
             callOptions,
             tableId,
             metrics,
-            timer);
+            timer,
+            userCallbackExecutor);
 
     return new TableAsync(base);
   }

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableBase.java

@@ -39,10 +39,12 @@
 import io.grpc.Context;
 import io.grpc.Deadline;
 import io.grpc.Metadata;
+import java.util.concurrent.Executor;
 
 class TableBase implements AutoCloseable {
   private final SessionPool<?> sessionPool;
   private final BigtableTimer timer;
+  private final Executor userCallbackExecutor;
   private final Metrics metrics;
   private final VRpcDescriptor<?, SessionReadRowRequest, SessionReadRowResponse> readRowDescriptor;
   private final VRpcDescriptor<?, SessionMutateRowRequest, SessionMutateRowResponse>
@@ -60,7 +62,8 @@ static <ReqT extends Message> TableBase createAndStart(
       CallOptions callOptions,
       String sessionPoolName,
       Metrics metrics,
-      BigtableTimer timer) {
+      BigtableTimer timer,
+      Executor userCallbackExecutor) {
 
     SessionPool<ReqT> sessionPool =
         new SessionPoolImpl<>(
@@ -76,7 +79,8 @@ static <ReqT extends Message> TableBase createAndStart(
 
     sessionPool.start(openReq, new Metadata());
 
-    return new TableBase(sessionPool, readRowDescriptor, mutateRowDescriptor, metrics, timer);
+    return new TableBase(
+        sessionPool, readRowDescriptor, mutateRowDescriptor, metrics, timer, userCallbackExecutor);
   }
 
   @VisibleForTesting
@@ -85,12 +89,14 @@ static <ReqT extends Message> TableBase createAndStart(
       VRpcDescriptor<?, SessionReadRowRequest, SessionReadRowResponse> readRowDescriptor,
       VRpcDescriptor<?, SessionMutateRowRequest, SessionMutateRowResponse> mutateRowDescriptor,
       Metrics metrics,
-      BigtableTimer timer) {
+      BigtableTimer timer,
+      Executor userCallbackExecutor) {
     this.sessionPool = sessionPool;
     this.readRowDescriptor = readRowDescriptor;
     this.mutateRowDescriptor = mutateRowDescriptor;
     this.metrics = metrics;
     this.timer = timer;
+    this.userCallbackExecutor = userCallbackExecutor;
   }
 
   @Override
@@ -112,7 +118,8 @@ public void readRow(
         new RetryingVRpc<>(() -> sessionPool.newCall(readRowDescriptor), timer);
     VRpcTracer tracer = metrics.newTableTracer(sessionPool.getInfo(), readRowDescriptor, deadline);
 
-    new VOperationImpl<>(retry, Context.current(), tracer, deadline, true).start(req, listener);
+    new VOperationImpl<>(retry, Context.current(), userCallbackExecutor, tracer, deadline, true)
+        .start(req, listener);
   }
 
   public void mutateRow(
@@ -125,7 +132,8 @@ public void mutateRow(
     VRpcTracer tracer =
         metrics.newTableTracer(sessionPool.getInfo(), mutateRowDescriptor, deadline);
 
-    new VOperationImpl<>(retry, Context.current(), tracer, deadline, idempotent)
+    new VOperationImpl<>(
+            retry, Context.current(), userCallbackExecutor, tracer, deadline, idempotent)
         .start(req, listener);
   }
 }

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

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/compat/ShimImpl.java

@@ -47,13 +47,16 @@
 import com.google.cloud.bigtable.data.v2.stub.MetadataExtractorInterceptor;
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import io.grpc.ClientInterceptor;
 import io.grpc.ManagedChannel;
 import io.grpc.Metadata;
 import io.grpc.stub.MetadataUtils;
 import java.io.IOException;
 import java.time.Duration;
 import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
@@ -160,14 +163,22 @@ public static Shim create(
       featureFlags = featureFlags.toBuilder().setSessionsRequired(true).build();
     }
 
+    ExecutorService userCallbackExecutor =
+        Executors.newCachedThreadPool(
+            new ThreadFactoryBuilder()
+                .setNameFormat("bigtable-callback-shim-%d")
+                .setDaemon(true)
+                .build());
+
     Client client =
         new Client(
             clientChannelProvider.updateFeatureFlags(featureFlags),
             clientInfo,
             clientChannelProvider,
             Resource.createShared(metrics),
             Resource.createShared(configManager),
-            Resource.createShared(bgExecutor));
+            Resource.createShared(bgExecutor),
+            Resource.createOwned(userCallbackExecutor, userCallbackExecutor::shutdown));
 
     return new ShimImpl(configManager, client);
   }