chore: back OpExecutor with SerializingExecutor(userCallbackExecutor)

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.

Author: igorbernstein2

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);
   }
 }

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}).
  *
- * <p>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.
+ * <p>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;
           }

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<ReqT, RespT> implements VOperation<ReqT, RespT> {
 
   private final VRpc<ReqT, RespT> 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<ReqT, RespT> implements VOperation<ReqT, RespT> {
   public VOperationImpl(
       VRpc<ReqT, RespT> 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<RespT> 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));

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/middleware/VRpc.java

@@ -111,7 +111,10 @@ abstract class VRpcCallContext {
     public static VRpcCallContext create(
         Deadline deadline, boolean isIdempotent, VRpcTracer tracer) {
       return create(
-          deadline, isIdempotent, tracer, new OpExecutor(MoreExecutors.directExecutor()));
+          deadline,
+          isIdempotent,
+          tracer,
+          new OpExecutor(MoreExecutors.directExecutor(), t -> {}));
     }
 
     public static VRpcCallContext create(