chore: replace OpExecutor backing with an inline-capable queue; tighten Client.close

OpExecutor switches from a SequentialExecutor backing to an internal
ArrayDeque + drain loop, and gains runInline() — runs the task synchronously
on the caller thread when the executor is idle, otherwise queues it.
VOperationImpl uses runInline for chain.start so the start dispatch skips
the queue+drain round-trip. Drain rejections (e.g. during shutdown) reset
drainScheduled so subsequent submissions can retry.

Client.close drains userCallbackExecutor first via a 5s-bounded
shutdownAndAwait so pool.close's cancelWithResult onClose notifications
complete before the executor is torn down. Resource.close becomes idempotent.

Drive-by cleanups: replace fully-qualified type names with imports across
touched files, and swap Guava Objects.hashCode(id) for Long.hashCode(id) in
ChannelPoolDpImpl.AfeId to avoid per-call boxing and array allocation.

Author: igorbernstein2

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()

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/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()

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()

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/compat/ShimImpl.java

@@ -47,13 +47,16 @@
 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;
 import io.grpc.stub.MetadataUtils;
 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());

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

@@ -16,17 +16,26 @@
 
 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;
 
 /**
- * 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 +45,11 @@ public interface UncaughtExceptionHandler {
 
   private final Executor backing;
   private final UncaughtExceptionHandler handler;
+
+  // Guards queue and drainScheduled. runningThread is volatile so throwIfNotInThisExecutor() can
+  // read it lock-free; writes happen inside the lock to piggy-back the memory barrier.
+  private final ArrayDeque<Runnable> queue = new ArrayDeque<>();
+  private boolean drainScheduled = false;
   private volatile Thread runningThread;
 
   public OpExecutor(Executor backing, UncaughtExceptionHandler handler) {
@@ -45,18 +59,82 @@ 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.
+  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() {

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

@@ -86,18 +86,24 @@ public void start(ReqT req, VRpcListener<RespT> listener) {
     // single Done.
     OpExecutor exec =
         new OpExecutor(
-            MoreExecutors.newSequentialExecutor(userCallbackExecutor),
-            t -> chain.cancel("Uncaught exception in op executor task", t));
+            userCallbackExecutor, t -> chain.cancel("Uncaught exception in op executor task", t));
     this.opExecutor = exec;
     VRpcCallContext ctx = VRpcCallContext.create(deadline, idempotent, tracer, exec);
     CleanupListener<RespT> wrapped =
         new CleanupListener<>(listener, grpcContext, cancellationListener);
-    // Register the gRPC context listener BEFORE submitting chain.start. The submit queues the
-    // task on the op executor; chain.cancel from this listener also queues. SequentialExecutor
-    // preserves submission order, so a context-cancel fired during/before chain.start will be
-    // processed after it.
-    grpcContext.addListener(cancellationListener, MoreExecutors.directExecutor());
-    exec.execute(() -> chain.start(req, ctx, wrapped));
+    exec.runInline(() -> chain.start(req, ctx, wrapped));
+    // Register AFTER chain.start so a context-cancel that fires immediately is sequenced behind
+    // start. runInline runs chain.start synchronously, so it has fully completed by the time the
+    // listener is registered. Matches ClientCallImpl's ordering (grpc-java issue #1343).
+    //
+    // If the chain reached a terminal onClose synchronously inside runInline (uncaught-handler
+    // recovery, immediate failure), CleanupListener already tried to remove a listener that was
+    // never registered (no-op). Skip the addListener in that case — otherwise we'd register a
+    // listener on grpcContext that nothing will ever remove, pinning the entire chain for the
+    // lifetime of the (potentially long-lived) gRPC Context.
+    if (!wrapped.closed) {
+      grpcContext.addListener(cancellationListener, MoreExecutors.directExecutor());
+    }
   }
 
   @Override

java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/session/SessionImplTest.java

@@ -70,11 +70,13 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -679,9 +681,9 @@ void abortFiresWhenListenerOnReadyThrows() throws Exception {
     SessionImpl session =
         new SessionImpl(metrics, poolInfo, 0, sessionFactory.createNew(), timer);
 
-    java.util.concurrent.CountDownLatch onCloseLatch = new java.util.concurrent.CountDownLatch(1);
-    java.util.concurrent.atomic.AtomicReference<Status> capturedStatus =
-        new java.util.concurrent.atomic.AtomicReference<>();
+    CountDownLatch onCloseLatch = new CountDownLatch(1);
+    AtomicReference<Status> capturedStatus =
+        new AtomicReference<>();
 
     Session.Listener throwingListener =
         new Session.Listener() {
@@ -721,8 +723,8 @@ void abortDoesNotHangWhenListenerOnCloseThrows() throws Exception {
     SessionImpl session =
         new SessionImpl(metrics, poolInfo, 0, sessionFactory.createNew(), timer);
 
-    java.util.concurrent.CountDownLatch onReadyLatch = new java.util.concurrent.CountDownLatch(1);
-    java.util.concurrent.CountDownLatch onCloseLatch = new java.util.concurrent.CountDownLatch(1);
+    CountDownLatch onReadyLatch = new CountDownLatch(1);
+    CountDownLatch onCloseLatch = new CountDownLatch(1);
 
     Session.Listener throwingListener =
         new Session.Listener() {
@@ -775,8 +777,8 @@ void abortDoesNotInfiniteLoopWhenRecoveryListenerAlsoThrows() throws Exception {
     SessionImpl session =
         new SessionImpl(metrics, poolInfo, 0, sessionFactory.createNew(), timer);
 
-    java.util.concurrent.CountDownLatch onCloseInvoked =
-        new java.util.concurrent.CountDownLatch(1);
+    CountDownLatch onCloseInvoked =
+        new CountDownLatch(1);
 
     Session.Listener doublyThrowingListener =
         new Session.Listener() {