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..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 { @@ -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); } 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..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 { @@ -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); } 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..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; @@ -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); } 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..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 @@ -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 @@ -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); } } 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 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..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,6 +163,13 @@ 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), @@ -167,7 +177,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/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..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 @@ -110,7 +110,8 @@ 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( 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..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; @@ -72,7 +73,8 @@ public void setup() { VRpcDescriptor.READ_ROW, VRpcDescriptor.MUTATE_ROW, noopMetrics, - mockTimer); + mockTimer, + MoreExecutors.directExecutor()); deadline = Deadline.after(1, TimeUnit.MINUTES); f = new UnaryResponseFuture<>(); }