feat(bigtable): BigtableDataClientFactory session support

This commit extends the session-based API path to support BigtableDataClientFactory.

Now the factory creates a single shared ChannelPool and
  ClientConfigurationManager once and hands lightweight child clients a reference to both.

  The bulk of the implementation work is a rewrite of ChannelPoolDpImpl to correctly handle multiple tenants (different instances/app profiles) sharing a single channel pool.

  ---
  ChannelPoolDpImpl: New Placement Model

  Old design

  The old pool organized channels into AfeChannelGroup objects — one deque of channels per observed AFE. When a channel's first session revealed its AFE (via onBeforeSessionStart), the channel was rehomed from
  the temporary startingGroup into the matching AfeChannelGroup, creating one if it didn't exist. New streams were then placed by picking the group with the fewest sessions.

  This design had a problem for the multi-tenant factory case:
  AFE affinity was tracked per channel, but for factory children different tenants on the same channel can be routed to different AFEs by RLS. A channel in group "AFE-7" might route tenant A to AFE-7 but
  tenant B somewhere else entirely.

  New design

  The new pool replaces the nested group structure with a flat channels list plus a routeObservations map keyed on (channelId, tenantKey) → lastObservedAfeId.

  Data structures:
  - channels: List<ChannelWrapper> — all channels in ACTIVE or DRAINING state
  - routeObservations: Map<RouteKey, AfeId> — the last AFE seen for a given (channel, tenant) pair, populated on every onBeforeSessionStart and self-healing as RLS routes drift over time
  - sessionsPerAfeId: Multiset<AfeId> — global count of live sessions per AFE, used to find underloaded targets
  - TenantKey (new value type) — stamped onto CallOptions by TableBase so the pool can distinguish tenants within the same shared pool

  Placement (newStream):

  1. Capacity mode — find the AFE with the smallest session count that is still below softMaxPerGroup:
    - 3a — Preferred: pick the ACTIVE channel whose last observed route for this tenant was that AFE, choosing the least-loaded among candidates. This is the steady-state path: RLS is stable per (channel,
  tenant), so repeat sessions for the same tenant land on the same AFE.
    - 3b — Unobserved-tenant fallback: if no route observation exists for this tenant on any channel, pick any ACTIVE channel with remaining capacity (least-loaded). The actual AFE is unknown until
  onBeforeSessionStart fires; routeObservations is updated then for future placements.
  2. Diversity mode — entered when all known AFEs are at cap, or the pool is empty: prefer an ACTIVE channel with fewer than softMaxPerGroup / 2 outstanding streams. If none exists, open a new channel and
  absorb whichever AFE it lands on.

  Scale-in (DRAINING state):

  The old "thinning" logic would call channel.shutdown() during serviceChannels(). The new approach introduces a two-state channel lifecycle
  (ACTIVE / DRAINING). When serviceChannels() decides to shrink the pool it marks excess channels DRAINING — they stop accepting new streams but continue serving existing ones. A DRAINING channel is removed
  only when its numOutstanding reaches zero (in releaseChannel). Drain candidates are chosen by pickDrainCandidates, which prefers idle channels first, then least-recently-used, then oldest.

  Route observation cleanup:

  removeChannel purges all routeObservations entries for the removed channel's ID, preventing the map from growing unboundedly as channels cycle.

  ---
  Other changes

  BigtableDataClientFactory / BigtableClientContext:
  - BigtableDataClientFactory.create() now calls BigtableClientContext.createForFactory() (a new entry point) instead of manually disabling sessions. The factory context initializes a shared ShimImpl (channel
  pool + config manager) that is reused by all children.
  - BigtableClientContext.createChild() now creates a lightweight ShimImpl child via ShimImpl.createForFactoryChild() rather than sharing a DisabledShim.

  ShimImpl / Client:
  - Client.channelPool changed from a bare ChannelPool reference to Resource<ChannelPool> so the factory-child constructor can hold a shared (non-closing) reference and the standard constructor holds an owned
  one.
  - ShimImpl.configManager similarly wrapped in Resource<> so close() is a no-op for factory children (the parent owns the manager's lifecycle).

Author: nimf

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/BigtableDataClientFactory.java

@@ -74,12 +74,8 @@ public final class BigtableDataClientFactory implements AutoCloseable {
    */
   public static BigtableDataClientFactory create(BigtableDataSettings defaultSettings)
       throws IOException {
-    BigtableDataSettings.Builder builder = defaultSettings.toBuilder();
-    builder.stubSettings().setSessionsEnabled(false);
-    defaultSettings = builder.build();
-
     BigtableClientContext sharedClientContext =
-        BigtableClientContext.create(defaultSettings.getStubSettings());
+        BigtableClientContext.createForFactory(defaultSettings.getStubSettings());
     ClientOperationSettings perOpSettings = defaultSettings.getStubSettings().getPerOpSettings();
     return new BigtableDataClientFactory(sharedClientContext, perOpSettings);
   }

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

@@ -73,7 +73,7 @@ public class Client implements AutoCloseable {
   private final Resource<ScheduledExecutorService> backgroundExecutor;
 
   private final CallOptions defaultCallOptions;
-  private final ChannelPool channelPool;
+  private final Resource<ChannelPool> channelPool;
   private final Resource<Metrics> metrics;
   private final Resource<ClientConfigurationManager> configManager;
 
@@ -147,19 +147,23 @@ public static Client create(ClientSettings settings) throws IOException {
     return new Client(
         featureFlags,
         clientInfo,
-        settings.getChannelProvider(),
         Resource.createOwned(metrics, metrics::close),
         Resource.createOwned(configManager, configManager::close),
-        Resource.createOwned(backgroundExecutor, backgroundExecutor::shutdown));
+        Resource.createOwned(backgroundExecutor, backgroundExecutor::shutdown),
+        settings.getChannelProvider());
   }
 
+  /**
+   * Standard constructor used by non-factory clients. Builds an owned {@link SwitchingChannelPool}
+   * from the provided {@link ChannelProvider}.
+   */
   public Client(
       FeatureFlags featureFlags,
       ClientInfo clientInfo,
-      ChannelProvider channelProvider,
       Resource<Metrics> metrics,
       Resource<ClientConfigurationManager> configManager,
-      Resource<ScheduledExecutorService> bgExecutor)
+      Resource<ScheduledExecutorService> bgExecutor,
+      ChannelProvider channelProvider)
       throws IOException {
     this.featureFlags = featureFlags;
     this.clientInfo = clientInfo;
@@ -181,13 +185,34 @@ public Client(
                     // TODO: consider localizing this for large reads
                     .maxInboundMessageSize(256 * 1024 * 1024));
 
-    channelPool =
+    SwitchingChannelPool switchingPool =
         new SwitchingChannelPool(
             configuredChannelProvider,
             configManager.get(),
             metrics.get(),
             backgroundExecutor.get());
-    channelPool.start();
+    switchingPool.start();
+    this.channelPool = Resource.createOwned(switchingPool, switchingPool::close);
+  }
+
+  /**
+   * Factory-child constructor. Uses a pre-built, shared {@link ChannelPool} and {@link
+   * ClientConfigurationManager}. The pool is already started and must not be closed by this client.
+   */
+  public Client(
+      FeatureFlags featureFlags,
+      ClientInfo clientInfo,
+      Resource<Metrics> metrics,
+      Resource<ClientConfigurationManager> configManager,
+      Resource<ScheduledExecutorService> bgExecutor,
+      Resource<ChannelPool> sharedChannelPool) {
+    this.featureFlags = featureFlags;
+    this.clientInfo = clientInfo;
+    this.metrics = metrics;
+    this.configManager = configManager;
+    this.backgroundExecutor = bgExecutor;
+    this.channelPool = sharedChannelPool;
+    defaultCallOptions = CallOptions.DEFAULT;
   }
 
   @Override
@@ -200,7 +225,7 @@ public void close() {
                     .setDescription("Client closing")
                     .build()));
     metrics.close();
-    channelPool.close();
+    channelPool.close(); // no-op when Resource.createShared (factory child)
     configManager.close();
     backgroundExecutor.close();
   }
@@ -211,7 +236,7 @@ public TableAsync openTableAsync(String tableId, Permission permission) {
             featureFlags,
             clientInfo,
             configManager.get(),
-            channelPool,
+            channelPool.get(),
             defaultCallOptions,
             tableId,
             permission,
@@ -228,7 +253,7 @@ public AuthorizedViewAsync openAuthorizedViewAsync(
             featureFlags,
             clientInfo,
             configManager.get(),
-            channelPool,
+            channelPool.get(),
             defaultCallOptions,
             tableId,
             viewId,
@@ -246,7 +271,7 @@ public MaterializedViewAsync openMaterializedViewAsync(
             featureFlags,
             clientInfo,
             configManager.get(),
-            channelPool,
+            channelPool.get(),
             defaultCallOptions,
             viewId,
             permission,
@@ -256,6 +281,16 @@ public MaterializedViewAsync openMaterializedViewAsync(
     return viewAsync;
   }
 
+  /** Returns the underlying channel pool (e.g. for sharing with factory children). */
+  public ChannelPool getChannelPool() {
+    return channelPool.get();
+  }
+
+  /** Returns the feature flags (e.g. for sharing with factory children). */
+  public FeatureFlags getFeatureFlags() {
+    return featureFlags;
+  }
+
   public static class Resource<T> {
     private T value;
     private Runnable closer;

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

@@ -22,6 +22,8 @@
 import com.google.bigtable.v2.SessionReadRowRequest;
 import com.google.bigtable.v2.SessionReadRowResponse;
 import com.google.cloud.bigtable.data.v2.internal.channels.ChannelPool;
+import com.google.cloud.bigtable.data.v2.internal.channels.ChannelPoolOptions;
+import com.google.cloud.bigtable.data.v2.internal.channels.TenantKey;
 import com.google.cloud.bigtable.data.v2.internal.csm.Metrics;
 import com.google.cloud.bigtable.data.v2.internal.csm.attributes.ClientInfo;
 import com.google.cloud.bigtable.data.v2.internal.csm.tracers.VRpcTracer;
@@ -63,14 +65,20 @@ static <ReqT extends Message> TableBase createAndStart(
       Metrics metrics,
       ScheduledExecutorService executor) {
 
+    // Stamp the tenant key so ChannelPoolDpImpl can make tenant-aware placement decisions.
+    CallOptions stamped =
+        callOptions.withOption(
+            ChannelPoolOptions.TENANT_KEY_OPTION,
+            new TenantKey(clientInfo.getInstanceName(), clientInfo.getAppProfileId()));
+
     SessionPool<ReqT> sessionPool =
         new SessionPoolImpl<>(
             metrics,
             featureFlags,
             clientInfo,
             configManager,
             channelPool,
-            callOptions,
+            stamped,
             sessionDescriptor,
             sessionPoolName,
             executor);