Threading refactor (4/7): Pool & op-executor contract

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

@@ -171,7 +171,7 @@ public static Client create(ClientSettings settings) throws IOException {
         Resource.createOwned(metrics, metrics::close),
         Resource.createOwned(configManager, configManager::close),
         Resource.createOwned(backgroundExecutor, backgroundExecutor::shutdown),
-        Resource.createOwned(userCallbackExecutor, userCallbackExecutor::shutdown));
+        Resource.createOwned(userCallbackExecutor, () -> shutdownAndAwait(userCallbackExecutor)));
   }
 
   public Client(
@@ -225,13 +225,17 @@ public void close() {
                     .setReason(CloseSessionReason.CLOSE_SESSION_REASON_USER)
                     .setDescription("Client closing")
                     .build()));
+    // Drain user-callback first so pool.close's cancelWithResult listener notifications complete
+    // before we tear down the surrounding executors and timer. Without this, the late onClose
+    // submissions race the shutdown and get RejectedExecutionException, silently dropping the
+    // user's terminal onClose.
+    userCallbackExecutor.close();
     metrics.close();
     channelPool.close();
     configManager.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) {
@@ -289,8 +293,10 @@ public MaterializedViewAsync openMaterializedViewAsync(
   }
 
   public static class Resource<T> {
-    private T value;
-    private Runnable closer;
+    private final T value;
+    private final Runnable closer;
+    private final java.util.concurrent.atomic.AtomicBoolean closed =
+        new java.util.concurrent.atomic.AtomicBoolean(false);
 
     public static <T> Resource<T> createOwned(T value, Runnable closer) {
       return new Resource<>(value, closer);
@@ -305,12 +311,29 @@ private Resource(T value, Runnable closer) {
       this.closer = closer;
     }
 
+    /** Idempotent. Repeat calls are no-ops. */
     public void close() {
-      this.closer.run();
+      if (closed.compareAndSet(false, true)) {
+        this.closer.run();
+      }
     }
 
     public T get() {
       return value;
     }
   }
+
+  // Drain in-flight listener.onClose tasks before the executor is shut down; bound the wait at 5s
+  // so close() doesn't hang the caller on a pathological listener.
+  private static void shutdownAndAwait(ExecutorService exec) {
+    exec.shutdown();
+    try {
+      if (!exec.awaitTermination(5, TimeUnit.SECONDS)) {
+        exec.shutdownNow();
+      }
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      exec.shutdownNow();
+    }
+  }
 }

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/ChannelPoolDpImpl.java

@@ -573,7 +573,7 @@ public boolean equals(Object o) {
 
     @Override
     public int hashCode() {
-      return com.google.common.base.Objects.hashCode(id);
+      return Long.hashCode(id);
     }
 
     @Override

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

@@ -16,17 +16,27 @@
 
 package com.google.cloud.bigtable.data.v2.internal.middleware;
 
+import com.google.common.util.concurrent.MoreExecutors;
+import java.util.ArrayDeque;
 import java.util.concurrent.Executor;
+import javax.annotation.concurrent.GuardedBy;
 
 /**
- * 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
+ * Per-op serializing executor that tracks which thread is currently draining it, so callers can
+ * assert they are running inside the executor (analogous to {@code
  * SynchronizationContext#throwIfNotInThisSynchronizationContext}).
  *
- * <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.
+ * <p>{@link #runInline} executes synchronously on the caller thread when the executor is idle, or
+ * falls back to a queued submission when busy. Use it to avoid the queue+drain round-trip for the
+ * first task of a fresh op executor (e.g. the start() dispatch).
+ *
+ * <p>If a task throws, the registered {@link UncaughtExceptionHandler} is invoked on the same task
+ * thread (still inside the drain wrapper, so {@link #throwIfNotInThisExecutor} still passes). This
+ * is the last-resort recovery point for the op chain — without it, an exception inside a callback
+ * is silently dropped and the caller's listener never sees a terminal close. The handler's own
+ * throws propagate (caught by the backing executor in production; surfaced to the calling thread
+ * when the backing is {@link MoreExecutors#directExecutor()}, which is what makes fail-fast test
+ * handlers like {@code t -> throw new AssertionError(t)} work).
  */
 public final class OpExecutor implements Executor {
 
@@ -36,6 +46,15 @@ public interface UncaughtExceptionHandler {
 
   private final Executor backing;
   private final UncaughtExceptionHandler handler;
+
+  // queue is the lock. runningThread is volatile (not @GuardedBy): throwIfNotInThisExecutor()
+  // reads it lock-free; writes happen inside synchronized(queue) to piggy-back the memory barrier.
+  @GuardedBy("queue")
+  private final ArrayDeque<Runnable> queue = new ArrayDeque<>();
+
+  @GuardedBy("queue")
+  private boolean drainScheduled = false;
+
   private volatile Thread runningThread;
 
   public OpExecutor(Executor backing, UncaughtExceptionHandler handler) {
@@ -45,18 +64,84 @@ public OpExecutor(Executor backing, UncaughtExceptionHandler handler) {
 
   @Override
   public void execute(Runnable r) {
-    backing.execute(
-        () -> {
-          Thread prev = runningThread;
-          runningThread = Thread.currentThread();
-          try {
-            r.run();
-          } catch (Throwable t) {
-            handler.uncaught(t);
-          } finally {
-            runningThread = prev;
-          }
-        });
+    synchronized (queue) {
+      queue.add(r);
+      if (!drainScheduled && runningThread == null) {
+        scheduleDrainLocked();
+      }
+    }
+  }
+
+  /**
+   * Runs {@code r} synchronously on the caller thread if this executor is idle, otherwise queues it
+   * for later drain on the backing executor. Either way, FIFO ordering with other tasks is
+   * preserved.
+   */
+  public void runInline(Runnable r) {
+    synchronized (queue) {
+      if (drainScheduled || runningThread != null || !queue.isEmpty()) {
+        queue.add(r);
+        if (!drainScheduled) {
+          scheduleDrainLocked();
+        }
+        return;
+      }
+      runningThread = Thread.currentThread();
+    }
+    try {
+      try {
+        r.run();
+      } catch (Throwable t) {
+        handler.uncaught(t);
+      }
+    } finally {
+      synchronized (queue) {
+        runningThread = null;
+        if (!queue.isEmpty() && !drainScheduled) {
+          scheduleDrainLocked();
+        }
+      }
+    }
+  }
+
+  // Schedule a drain on the backing executor. If the backing throws (e.g.
+  // RejectedExecutionException
+  // during shutdown), reset drainScheduled before propagating so the next execute() can retry
+  // instead of wedging the executor with no drainer.
+  @GuardedBy("queue")
+  private void scheduleDrainLocked() {
+    drainScheduled = true;
+    try {
+      backing.execute(this::drain);
+    } catch (Throwable t) {
+      drainScheduled = false;
+      throw t;
+    }
+  }
+
+  private void drain() {
+    while (true) {
+      Runnable r;
+      synchronized (queue) {
+        r = queue.poll();
+        if (r == null) {
+          drainScheduled = false;
+          return;
+        }
+        runningThread = Thread.currentThread();
+      }
+      try {
+        try {
+          r.run();
+        } catch (Throwable t) {
+          handler.uncaught(t);
+        }
+      } finally {
+        synchronized (queue) {
+          runningThread = null;
+        }
+      }
+    }
   }
 
   public void throwIfNotInThisExecutor() {