Threading refactor (7/8): replace shaded-netty HashedWheelTimer with in-process port

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

@@ -50,6 +50,7 @@ static AuthorizedViewAsync createAndStart(
       Permission permission,
       Metrics metrics,
       BigtableTimer timer,
+      Executor backgroundExecutor,
       Executor userCallbackExecutor) {
 
     AuthorizedViewName viewName =
@@ -81,6 +82,7 @@ static AuthorizedViewAsync createAndStart(
             viewName.toString(),
             metrics,
             timer,
+            backgroundExecutor,
             userCallbackExecutor);
 
     return new AuthorizedViewAsync(base);

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

@@ -34,7 +34,7 @@
 import com.google.cloud.bigtable.data.v2.internal.csm.NoopMetrics;
 import com.google.cloud.bigtable.data.v2.internal.csm.attributes.ClientInfo;
 import com.google.cloud.bigtable.data.v2.internal.session.BigtableTimer;
-import com.google.cloud.bigtable.data.v2.internal.session.NettyWheelTimer;
+import com.google.cloud.bigtable.data.v2.internal.session.HashedWheelTimer;
 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;
@@ -205,9 +205,7 @@ public Client(
     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());
+    this.sessionTimer = new HashedWheelTimer("bigtable-session-timer");
 
     defaultCallOptions = CallOptions.DEFAULT;
 
@@ -311,6 +309,7 @@ public TableAsync openTableAsync(String tableId, Permission permission) {
               permission,
               metrics.get(),
               sessionTimer,
+              backgroundExecutor.get(),
               userCallbackExecutor.get());
       sessionPools.add(tableAsync.getSessionPool());
       return tableAsync;
@@ -335,6 +334,7 @@ public AuthorizedViewAsync openAuthorizedViewAsync(
               permission,
               metrics.get(),
               sessionTimer,
+              backgroundExecutor.get(),
               userCallbackExecutor.get());
       sessionPools.add(viewAsync.getSessionPool());
       return viewAsync;
@@ -358,6 +358,7 @@ public MaterializedViewAsync openMaterializedViewAsync(
               permission,
               metrics.get(),
               sessionTimer,
+              backgroundExecutor.get(),
               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

@@ -46,6 +46,7 @@ public static MaterializedViewAsync createAndStart(
       OpenMaterializedViewRequest.Permission permission,
       Metrics metrics,
       BigtableTimer timer,
+      Executor backgroundExecutor,
       Executor userCallbackExecutor) {
 
     MaterializedViewName viewName =
@@ -76,6 +77,7 @@ public static MaterializedViewAsync createAndStart(
             viewId,
             metrics,
             timer,
+            backgroundExecutor,
             userCallbackExecutor);
 
     return new MaterializedViewAsync(base);

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

@@ -49,6 +49,7 @@ public static TableAsync createAndStart(
       Permission permission,
       Metrics metrics,
       BigtableTimer timer,
+      Executor backgroundExecutor,
       Executor userCallbackExecutor) {
 
     TableName tableName =
@@ -79,6 +80,7 @@ public static TableAsync createAndStart(
             tableId,
             metrics,
             timer,
+            backgroundExecutor,
             userCallbackExecutor);
 
     return new TableAsync(base);

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

@@ -63,6 +63,7 @@ static <ReqT extends Message> TableBase createAndStart(
       String sessionPoolName,
       Metrics metrics,
       BigtableTimer timer,
+      Executor backgroundExecutor,
       Executor userCallbackExecutor) {
 
     SessionPool<ReqT> sessionPool =
@@ -75,7 +76,8 @@ static <ReqT extends Message> TableBase createAndStart(
             callOptions,
             sessionDescriptor,
             sessionPoolName,
-            timer);
+            timer,
+            backgroundExecutor);
 
     sessionPool.start(openReq, new Metadata());
 

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

@@ -326,36 +326,21 @@ class Scheduled extends State {
 
     @Override
     public void onStart() {
-      try {
-        stopHook = timer.onStop(this::onTimerStopping);
-        // Wraps go innermost so the captured gRPC + OpenTelemetry contexts are re-established at
-        // the moment the body runs, not just while the dispatcher is invoking the outer task.
-        future =
-            timer.newTimeout(
-                () ->
-                    context
-                        .getExecutor()
-                        .execute(
-                            () ->
-                                grpcContext
-                                    .wrap(
-                                        () ->
-                                            otelContext.wrap(() -> onStateChange(new Idle())).run())
-                                    .run()),
-                Durations.toMillis(retryDelay),
-                TimeUnit.MILLISECONDS);
-      } catch (IllegalStateException e) {
-        // Timer was stopped between Active.onClose deciding to retry and this task running on the
-        // op executor. Race window is narrow (post-drain shutdown), but cover it cleanly so the
-        // op-executor uncaught handler does not have to. onExit will release the stopHook.
-        onStateChange(
-            new Done(
-                VRpcResult.createRejectedError(
-                    Status.CANCELLED
-                        .withDescription(
-                            "Executor shutting down, can't schedule operation for retry.")
-                        .withCause(e))));
-      }
+      // If the timer is already stopped, onStop fires the hook synchronously here — which
+      // enqueues an onTimerStopping body on the op executor — and newTimeout returns a
+      // pre-cancelled Timeout. The queued onTimerStopping then drives us to Done(CANCELLED).
+      stopHook = timer.onStop(this::onTimerStopping);
+      // Wraps go innermost so the captured gRPC + OpenTelemetry contexts are re-established at
+      // the moment the body runs on the op executor.
+      future =
+          timer.newTimeout(
+              () ->
+                  grpcContext
+                      .wrap(() -> otelContext.wrap(() -> onStateChange(new Idle())).run())
+                      .run(),
+              context.getExecutor(),
+              Durations.toMillis(retryDelay),
+              TimeUnit.MILLISECONDS);
     }
 
     // Invoked from BigtableTimer.stop on the close thread. Trampoline back to the op executor so

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/session/BigtableTimer.java

@@ -15,35 +15,32 @@
  */
 package com.google.cloud.bigtable.data.v2.internal.session;
 
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
 /**
  * Schedules short-lived callbacks (heartbeat ticks, deadline monitors, watchdog ticks) at
  * approximate, low-resolution times. Backed by a hashed wheel: O(1) insert and O(1) cancel,
  * regardless of how many pending timeouts the wheel holds.
  *
- * <p>{@link #newTimeout} runs the callback on the timer's bundled dispatch executor — callers do
- * not have to wrap their bodies in {@code executor.execute(...)} to stay off the tick thread. This
- * is the default and is correct for any callback that takes a lock or does real work.
- *
- * <p>TODO: once later refactor steps introduce per-op / per-session dispatchers (e.g. {@code
- * SerializingExecutor} or {@code SynchronizationContext}), add a {@code newTimeoutOnTickThread}
- * variant so callers with their own dispatcher can skip the wasted hop through the bundled
- * executor.
- *
- * <p>This is a thin abstraction over a single concrete implementation today (see {@code
- * NettyWheelTimer}). It exists so the implementation can be swapped after benchmarking establishes
- * a baseline.
+ * <p>Each {@link #newTimeout} call carries the {@link Executor} that should run the task body.
+ * Tasks never run on the timer's tick thread (unless the caller passes {@code directExecutor()};
+ * see warning below). Passing the executor per-call avoids the dispatcher hop that a bundled
+ * default would impose on callers who already trampoline onto their own executor.
  */
 public interface BigtableTimer {
   /**
-   * Schedules {@code task} to run after {@code delay}. The task body runs on the timer's bundled
-   * dispatch executor, not on the tick thread, so it is safe to take locks or do bounded work.
+   * Schedules {@code task} to run after {@code delay}. The task body is handed to {@code
+   * executor.execute(task)} — pass the executor where this task should ultimately run.
    *
    * <p>The returned handle can be used to cancel the task; cancel is O(1) and does not leave the
    * entry in any heap.
+   *
+   * <p><b>Warning:</b> passing {@code MoreExecutors.directExecutor()} runs {@code task} inline on
+   * the timer's tick thread. The task must be trivial and non-blocking — anything more will stall
+   * every other scheduled timeout on the wheel.
    */
-  Timeout newTimeout(Runnable task, long delay, TimeUnit unit);
+  Timeout newTimeout(Runnable task, Executor executor, long delay, TimeUnit unit);
 
   /**
    * Releases the tick thread and discards any pending timeouts. Idempotent. After {@code stop()},