From fe991197158d81315e18c0106346f93d5fb5d3c8 Mon Sep 17 00:00:00 2001 From: Igor Bernstein Date: Sun, 14 Jun 2026 16:38:29 +0000 Subject: [PATCH 1/5] chore: isolate user-callback executor on a cached thread pool Client (and ShimImpl) own a dedicated bigtable-callback-%d cached pool, plumbed through *Async / TableBase. A blocked user callback can no longer starve heartbeats, retry delays, or pool bookkeeping (all of which run on backgroundExecutor). The op-level SerializingExecutor in a later commit will dispatch onto this pool. --- .../v2/internal/api/AuthorizedViewAsync.java | 6 ++-- .../bigtable/data/v2/internal/api/Client.java | 31 ++++++++++++++++--- .../internal/api/MaterializedViewAsync.java | 6 ++-- .../data/v2/internal/api/TableAsync.java | 6 ++-- .../data/v2/internal/api/TableBase.java | 12 +++++-- .../data/v2/internal/compat/ShimImpl.java | 10 +++++- .../data/v2/internal/api/TableBaseTest.java | 3 +- 7 files changed, 58 insertions(+), 16 deletions(-) diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/AuthorizedViewAsync.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/AuthorizedViewAsync.java index 2c4bb29f1c0e..59c6ebcbccae 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/AuthorizedViewAsync.java +++ b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/AuthorizedViewAsync.java @@ -48,7 +48,8 @@ static AuthorizedViewAsync createAndStart( String viewId, Permission permission, Metrics metrics, - BigtableTimer timer) { + BigtableTimer timer, + java.util.concurrent.Executor userCallbackExecutor) { AuthorizedViewName viewName = AuthorizedViewName.builder() @@ -78,7 +79,8 @@ static AuthorizedViewAsync createAndStart( callOptions, viewName.toString(), metrics, - timer); + timer, + userCallbackExecutor); return new AuthorizedViewAsync(base); } diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/Client.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/Client.java index 6cb909d96b95..0d63a07e866e 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/Client.java +++ b/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 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 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, Resource configManager, - Resource bgExecutor) + Resource bgExecutor, + Resource 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; } diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/MaterializedViewAsync.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/MaterializedViewAsync.java index e83c2acde29e..4ccce31cd714 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/MaterializedViewAsync.java +++ b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/MaterializedViewAsync.java @@ -44,7 +44,8 @@ public static MaterializedViewAsync createAndStart( String viewId, OpenMaterializedViewRequest.Permission permission, Metrics metrics, - BigtableTimer timer) { + BigtableTimer timer, + java.util.concurrent.Executor userCallbackExecutor) { MaterializedViewName viewName = MaterializedViewName.builder() @@ -73,7 +74,8 @@ public static MaterializedViewAsync createAndStart( callOptions, viewId, metrics, - timer); + timer, + userCallbackExecutor); return new MaterializedViewAsync(base); } diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableAsync.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableAsync.java index a33dd459b6e3..9f9384f0f86e 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableAsync.java +++ b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableAsync.java @@ -47,7 +47,8 @@ public static TableAsync createAndStart( String tableId, Permission permission, Metrics metrics, - BigtableTimer timer) { + BigtableTimer timer, + java.util.concurrent.Executor userCallbackExecutor) { TableName tableName = TableName.builder() @@ -76,7 +77,8 @@ public static TableAsync createAndStart( callOptions, tableId, metrics, - timer); + timer, + userCallbackExecutor); return new TableAsync(base); } diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableBase.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableBase.java index 590f46b47b99..9a4b0c322bdd 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableBase.java +++ b/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 readRowDescriptor; private final VRpcDescriptor @@ -60,7 +62,8 @@ static TableBase createAndStart( CallOptions callOptions, String sessionPoolName, Metrics metrics, - BigtableTimer timer) { + BigtableTimer timer, + Executor userCallbackExecutor) { SessionPool sessionPool = new SessionPoolImpl<>( @@ -76,7 +79,8 @@ static 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 TableBase createAndStart( VRpcDescriptor readRowDescriptor, VRpcDescriptor 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 diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/compat/ShimImpl.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/compat/ShimImpl.java index 353973dc8e58..fcddf47f03b2 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/compat/ShimImpl.java +++ b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/compat/ShimImpl.java @@ -160,6 +160,13 @@ public static Shim create( featureFlags = featureFlags.toBuilder().setSessionsRequired(true).build(); } + java.util.concurrent.ExecutorService userCallbackExecutor = + java.util.concurrent.Executors.newCachedThreadPool( + new com.google.common.util.concurrent.ThreadFactoryBuilder() + .setNameFormat("bigtable-callback-shim-%d") + .setDaemon(true) + .build()); + Client client = new Client( clientChannelProvider.updateFeatureFlags(featureFlags), @@ -167,7 +174,8 @@ public static Shim create( clientChannelProvider, Resource.createShared(metrics), Resource.createShared(configManager), - Resource.createShared(bgExecutor)); + Resource.createShared(bgExecutor), + Resource.createOwned(userCallbackExecutor, userCallbackExecutor::shutdown)); return new ShimImpl(configManager, client); } diff --git a/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/api/TableBaseTest.java b/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/api/TableBaseTest.java index 089a78cb30b2..9075b27fb876 100644 --- a/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/api/TableBaseTest.java +++ b/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/api/TableBaseTest.java @@ -72,7 +72,8 @@ public void setup() { VRpcDescriptor.READ_ROW, VRpcDescriptor.MUTATE_ROW, noopMetrics, - mockTimer); + mockTimer, + com.google.common.util.concurrent.MoreExecutors.directExecutor()); deadline = Deadline.after(1, TimeUnit.MINUTES); f = new UnaryResponseFuture<>(); } From ec55e0490bc7511528187d748cc6ceffe364188a Mon Sep 17 00:00:00 2001 From: Igor Bernstein Date: Sun, 14 Jun 2026 16:45:28 +0000 Subject: [PATCH 2/5] chore: back OpExecutor with SerializingExecutor(userCallbackExecutor) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit VOperationImpl now constructs OpExecutor over a per-call SequentialExecutor on the shared userCallbackExecutor, replacing the per-call SynchronizationContext. OpExecutor gains an UncaughtExceptionHandler ctor arg — the safety net that the removed RetryingVRpc-owned SyncContext provided. The 3-arg VRpcCallContext.create defaults to a no-op handler for tests; production callers go through VOperationImpl. --- .../data/v2/internal/api/TableBase.java | 6 +++-- .../v2/internal/middleware/OpExecutor.java | 22 +++++++++++++------ .../internal/middleware/VOperationImpl.java | 19 ++++++++++------ .../data/v2/internal/middleware/VRpc.java | 6 ++++- 4 files changed, 36 insertions(+), 17 deletions(-) diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableBase.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableBase.java index 9a4b0c322bdd..a3b4a535b0d9 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableBase.java +++ b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableBase.java @@ -118,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( @@ -131,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); } } diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/OpExecutor.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/OpExecutor.java index da3dbe49e343..7622647bcc38 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/OpExecutor.java +++ b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/OpExecutor.java @@ -19,22 +19,28 @@ import java.util.concurrent.Executor; /** - * Per-op serializing executor. Wraps a delegate {@link Executor} and tracks which thread is - * currently running a task, so callers can assert they are on the executor (analogous to {@code + * Per-op serializing executor. Wraps a delegate {@link Executor} (typically a per-call {@code + * SerializingExecutor} over the user-callback pool) and tracks which thread is currently running a + * task, so callers can assert they are on the executor (analogous to {@code * SynchronizationContext#throwIfNotInThisSynchronizationContext}). * - *

Backing executor evolves over the refactor — for now it is the per-call {@link - * io.grpc.SynchronizationContext} that {@link VOperationImpl} constructs. Later commits swap it for - * a {@code SerializingExecutor} over the user-callback pool, and eventually a tailored inline - * queue. + *

If a task throws, the registered {@link UncaughtExceptionHandler} is invoked — this is the + * last-resort recovery point for the chain. Without it, a throw from a user-installed tracer or a + * listener callback would silently drop and the caller's future would never complete. */ public final class OpExecutor implements Executor { + public interface UncaughtExceptionHandler { + void uncaught(Throwable t); + } + private final Executor backing; + private final UncaughtExceptionHandler handler; private volatile Thread runningThread; - public OpExecutor(Executor backing) { + public OpExecutor(Executor backing, UncaughtExceptionHandler handler) { this.backing = backing; + this.handler = handler; } @Override @@ -45,6 +51,8 @@ public void execute(Runnable r) { runningThread = Thread.currentThread(); try { r.run(); + } catch (Throwable t) { + handler.uncaught(t); } finally { runningThread = prev; } diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VOperationImpl.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VOperationImpl.java index 4d8b7427d144..8e8a1554bea8 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VOperationImpl.java +++ b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VOperationImpl.java @@ -24,8 +24,8 @@ import com.google.common.util.concurrent.MoreExecutors; import io.grpc.Context; import io.grpc.Deadline; -import io.grpc.SynchronizationContext; import java.util.Optional; +import java.util.concurrent.Executor; import java.util.concurrent.TimeoutException; import javax.annotation.Nullable; @@ -39,6 +39,7 @@ public class VOperationImpl implements VOperation { private final VRpc chain; private final Context grpcContext; + private final Executor userCallbackExecutor; private final VRpcTracer tracer; private final Deadline deadline; private final boolean idempotent; @@ -47,11 +48,13 @@ public class VOperationImpl implements VOperation { public VOperationImpl( VRpc chain, Context grpcContext, + Executor userCallbackExecutor, VRpcTracer tracer, Deadline deadline, boolean idempotent) { this.chain = chain; this.grpcContext = grpcContext; + this.userCallbackExecutor = userCallbackExecutor; this.tracer = tracer; this.deadline = deadline; this.idempotent = idempotent; @@ -69,12 +72,14 @@ public VOperationImpl( @Override public void start(ReqT req, VRpcListener listener) { - // Per-call SynchronizationContext serializes all middleware below this layer. Uncaught task - // failures drive the chain to a terminal state so the caller's listener still gets onClose; - // RetryingVRpc.cancel is idempotent so the resulting cascade collapses safely. - SynchronizationContext syncContext = - new SynchronizationContext((t, e) -> chain.cancel("Uncaught exception in op executor", e)); - OpExecutor exec = new OpExecutor(syncContext); + // Per-call SerializingExecutor over the shared user-callback pool. The handler is the + // last-resort recovery: if any op-executor task throws (typically a user-installed tracer or + // a listener callback escape), drive the chain to a terminal state so the caller's listener + // still receives an onClose. RetryingVRpc.cancel is idempotent so cascades collapse safely. + OpExecutor exec = + new OpExecutor( + MoreExecutors.newSequentialExecutor(userCallbackExecutor), + t -> chain.cancel("Uncaught exception in op executor task", t)); VRpcCallContext ctx = VRpcCallContext.create(deadline, idempotent, tracer, exec); grpcContext.addListener(cancellationListener, MoreExecutors.directExecutor()); chain.start(req, ctx, new CleanupListener<>(listener, grpcContext, cancellationListener)); diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VRpc.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VRpc.java index 5d98f13bfdde..d5a605461d37 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VRpc.java +++ b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VRpc.java @@ -110,7 +110,11 @@ abstract class VRpcCallContext { */ public static VRpcCallContext create( Deadline deadline, boolean isIdempotent, VRpcTracer tracer) { - return create(deadline, isIdempotent, tracer, new OpExecutor(MoreExecutors.directExecutor())); + return create( + deadline, + isIdempotent, + tracer, + new OpExecutor(MoreExecutors.directExecutor(), t -> {})); } public static VRpcCallContext create( From b82e8c0efdb18573d1d7015aebe3ca8e77879588 Mon Sep 17 00:00:00 2001 From: Igor Bernstein Date: Thu, 28 May 2026 02:22:22 +0000 Subject: [PATCH 3/5] 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. --- .../data/v2/internal/channels/ChannelPoolDpImpl.java | 6 +++++- .../bigtable/data/v2/internal/channels/SessionStream.java | 8 ++++++++ .../data/v2/internal/channels/SingleChannelPool.java | 6 +++++- 3 files changed, 18 insertions(+), 2 deletions(-) diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/ChannelPoolDpImpl.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/ChannelPoolDpImpl.java index 7913b28ef14c..9a60a271ea7f 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/ChannelPoolDpImpl.java +++ b/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 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 diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/SessionStream.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/SessionStream.java index ebbc39af7f2c..eb6dbfa339eb 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/SessionStream.java +++ b/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. + * + *

Invariant: 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); diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/SingleChannelPool.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/SingleChannelPool.java index 6d40b58d53d4..bfd099f60d51 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/SingleChannelPool.java +++ b/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 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 From 4915191bcc714b97503c5e3c6adb6c8106685400 Mon Sep 17 00:00:00 2001 From: Igor Bernstein Date: Mon, 22 Jun 2026 19:21:08 +0000 Subject: [PATCH 4/5] chore: replace fully-qualified type names with imports --- .../data/v2/internal/api/AuthorizedViewAsync.java | 3 ++- .../data/v2/internal/api/MaterializedViewAsync.java | 3 ++- .../cloud/bigtable/data/v2/internal/api/TableAsync.java | 3 ++- .../cloud/bigtable/data/v2/internal/compat/ShimImpl.java | 9 ++++++--- .../bigtable/data/v2/internal/api/TableBaseTest.java | 3 ++- 5 files changed, 14 insertions(+), 7 deletions(-) diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/AuthorizedViewAsync.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/AuthorizedViewAsync.java index 59c6ebcbccae..adb6b2e2e0c0 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/AuthorizedViewAsync.java +++ b/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 { @@ -49,7 +50,7 @@ static AuthorizedViewAsync createAndStart( Permission permission, Metrics metrics, BigtableTimer timer, - java.util.concurrent.Executor userCallbackExecutor) { + Executor userCallbackExecutor) { AuthorizedViewName viewName = AuthorizedViewName.builder() diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/MaterializedViewAsync.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/MaterializedViewAsync.java index 4ccce31cd714..eb25b1d858dd 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/MaterializedViewAsync.java +++ b/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 { @@ -45,7 +46,7 @@ public static MaterializedViewAsync createAndStart( OpenMaterializedViewRequest.Permission permission, Metrics metrics, BigtableTimer timer, - java.util.concurrent.Executor userCallbackExecutor) { + Executor userCallbackExecutor) { MaterializedViewName viewName = MaterializedViewName.builder() diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableAsync.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableAsync.java index 9f9384f0f86e..8bb16c51c201 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/TableAsync.java +++ b/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; @@ -48,7 +49,7 @@ public static TableAsync createAndStart( Permission permission, Metrics metrics, BigtableTimer timer, - java.util.concurrent.Executor userCallbackExecutor) { + Executor userCallbackExecutor) { TableName tableName = TableName.builder() diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/compat/ShimImpl.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/compat/ShimImpl.java index fcddf47f03b2..d1c7080828e8 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/compat/ShimImpl.java +++ b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/compat/ShimImpl.java @@ -47,6 +47,7 @@ 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; @@ -54,6 +55,8 @@ 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,9 +163,9 @@ public static Shim create( featureFlags = featureFlags.toBuilder().setSessionsRequired(true).build(); } - java.util.concurrent.ExecutorService userCallbackExecutor = - java.util.concurrent.Executors.newCachedThreadPool( - new com.google.common.util.concurrent.ThreadFactoryBuilder() + ExecutorService userCallbackExecutor = + Executors.newCachedThreadPool( + new ThreadFactoryBuilder() .setNameFormat("bigtable-callback-shim-%d") .setDaemon(true) .build()); diff --git a/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/api/TableBaseTest.java b/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/api/TableBaseTest.java index 9075b27fb876..70f2b20f8922 100644 --- a/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/api/TableBaseTest.java +++ b/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/api/TableBaseTest.java @@ -31,6 +31,7 @@ import com.google.cloud.bigtable.data.v2.internal.session.SessionPool; import com.google.cloud.bigtable.data.v2.internal.session.SessionPoolInfo; import com.google.cloud.bigtable.data.v2.internal.session.VRpcDescriptor; +import com.google.common.util.concurrent.MoreExecutors; import com.google.protobuf.Message; import io.grpc.Deadline; import io.grpc.Metadata; @@ -73,7 +74,7 @@ public void setup() { VRpcDescriptor.MUTATE_ROW, noopMetrics, mockTimer, - com.google.common.util.concurrent.MoreExecutors.directExecutor()); + MoreExecutors.directExecutor()); deadline = Deadline.after(1, TimeUnit.MINUTES); f = new UnaryResponseFuture<>(); } From e4ca57f9a38643c309d0cf22b49c827d1ffaabec Mon Sep 17 00:00:00 2001 From: Igor Bernstein Date: Mon, 22 Jun 2026 21:24:05 +0000 Subject: [PATCH 5/5] chore: apply fmt-maven-plugin --- .../cloud/bigtable/data/v2/internal/middleware/VRpc.java | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VRpc.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VRpc.java index d5a605461d37..f45d523b5efd 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VRpc.java +++ b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VRpc.java @@ -111,10 +111,7 @@ abstract class VRpcCallContext { public static VRpcCallContext create( Deadline deadline, boolean isIdempotent, VRpcTracer tracer) { return create( - deadline, - isIdempotent, - tracer, - new OpExecutor(MoreExecutors.directExecutor(), t -> {})); + deadline, isIdempotent, tracer, new OpExecutor(MoreExecutors.directExecutor(), t -> {})); } public static VRpcCallContext create(