fix: serialize open*/close so racing opens cannot create orphan pools

openTableAsync / openAuthorizedViewAsync / openMaterializedViewAsync
read 'closed' lock-free, then constructed the pool, then inserted it
into sessionPools — close() could CAS closed=true and snapshot
sessionPools in between, leaving the new pool orphaned: never closed,
its callbacks landing on shut-down executors.

Hold sessionPools' monitor across the closed check, the construction,
and the insert. Opens are infrequent (typically once per table at app
startup) so the monitor cost is negligible.

Move close()'s closed flip inside the same monitor too. With every
access now under the lock, downgrade 'closed' from AtomicBoolean to a
plain boolean — the CAS provided no value over a plain read+write
under the lock.

Author: igorbernstein2

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

@@ -104,10 +104,10 @@ public class Client implements AutoCloseable {
   private final Resource<ClientConfigurationManager> configManager;
 
   private final Set<SessionPool<?>> sessionPools = Collections.newSetFromMap(new WeakHashMap<>());
-  // Set true at the start of close(); guards openTableAsync / openAuthorizedViewAsync /
-  // openMaterializedViewAsync so concurrent opens during shutdown don't create pools the close
-  // path won't see.
-  private final AtomicBoolean closed = new AtomicBoolean(false);
+  // Guarded by sessionPools' monitor: close() sets it before snapshotting the pool set, and the
+  // open* methods check it before adding a new pool, so a racing open cannot insert a pool that
+  // close() has already missed in its snapshot.
+  private boolean closed = false;
 
   public static Client create(ClientSettings settings) throws IOException {
     FeatureFlags featureFlags =
@@ -235,12 +235,12 @@ public Client(
 
   @Override
   public void close() {
-    if (!closed.compareAndSet(false, true)) {
-      return; // idempotent
-    }
-
     List<SessionPool<?>> toClose;
     synchronized (sessionPools) {
+      if (closed) {
+        return; // idempotent
+      }
+      closed = true;
       toClose = new ArrayList<>(sessionPools);
     }
 
@@ -292,67 +292,77 @@ public void close() {
     backgroundExecutor.close();
   }
 
-  private void checkNotClosed() {
-    if (closed.get()) {
-      throw new IllegalStateException("Client is closed");
-    }
-  }
-
+  // The closed check and pool insertion run under sessionPools' monitor so close() (which flips
+  // closed under the same monitor) cannot snapshot the pool set between our check and our insert.
+  // Opens are infrequent (typically once per table at app startup), so holding the monitor across
+  // createAndStart is acceptable.
   public TableAsync openTableAsync(String tableId, Permission permission) {
-    checkNotClosed();
-    TableAsync tableAsync =
-        TableAsync.createAndStart(
-            featureFlags,
-            clientInfo,
-            configManager.get(),
-            channelPool,
-            defaultCallOptions,
-            tableId,
-            permission,
-            metrics.get(),
-            sessionTimer,
-            userCallbackExecutor.get());
-    sessionPools.add(tableAsync.getSessionPool());
-    return tableAsync;
+    synchronized (sessionPools) {
+      if (closed) {
+        throw new IllegalStateException("Client is closed");
+      }
+      TableAsync tableAsync =
+          TableAsync.createAndStart(
+              featureFlags,
+              clientInfo,
+              configManager.get(),
+              channelPool,
+              defaultCallOptions,
+              tableId,
+              permission,
+              metrics.get(),
+              sessionTimer,
+              userCallbackExecutor.get());
+      sessionPools.add(tableAsync.getSessionPool());
+      return tableAsync;
+    }
   }
 
   public AuthorizedViewAsync openAuthorizedViewAsync(
       String tableId, String viewId, OpenAuthorizedViewRequest.Permission permission) {
-    checkNotClosed();
-    AuthorizedViewAsync viewAsync =
-        AuthorizedViewAsync.createAndStart(
-            featureFlags,
-            clientInfo,
-            configManager.get(),
-            channelPool,
-            defaultCallOptions,
-            tableId,
-            viewId,
-            permission,
-            metrics.get(),
-            sessionTimer,
-            userCallbackExecutor.get());
-    sessionPools.add(viewAsync.getSessionPool());
-    return viewAsync;
+    synchronized (sessionPools) {
+      if (closed) {
+        throw new IllegalStateException("Client is closed");
+      }
+      AuthorizedViewAsync viewAsync =
+          AuthorizedViewAsync.createAndStart(
+              featureFlags,
+              clientInfo,
+              configManager.get(),
+              channelPool,
+              defaultCallOptions,
+              tableId,
+              viewId,
+              permission,
+              metrics.get(),
+              sessionTimer,
+              userCallbackExecutor.get());
+      sessionPools.add(viewAsync.getSessionPool());
+      return viewAsync;
+    }
   }
 
   public MaterializedViewAsync openMaterializedViewAsync(
       String viewId, OpenMaterializedViewRequest.Permission permission) {
-    checkNotClosed();
-    MaterializedViewAsync viewAsync =
-        MaterializedViewAsync.createAndStart(
-            featureFlags,
-            clientInfo,
-            configManager.get(),
-            channelPool,
-            defaultCallOptions,
-            viewId,
-            permission,
-            metrics.get(),
-            sessionTimer,
-            userCallbackExecutor.get());
-    sessionPools.add(viewAsync.getSessionPool());
-    return viewAsync;
+    synchronized (sessionPools) {
+      if (closed) {
+        throw new IllegalStateException("Client is closed");
+      }
+      MaterializedViewAsync viewAsync =
+          MaterializedViewAsync.createAndStart(
+              featureFlags,
+              clientInfo,
+              configManager.get(),
+              channelPool,
+              defaultCallOptions,
+              viewId,
+              permission,
+              metrics.get(),
+              sessionTimer,
+              userCallbackExecutor.get());
+      sessionPools.add(viewAsync.getSessionPool());
+      return viewAsync;
+    }
   }
 
   public static class Resource<T> {