feat(bigtable): BigtableDataClientFactory session support (#13829)

This PR 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,10 +74,6 @@ 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());
     ClientOperationSettings perOpSettings = defaultSettings.getStubSettings().getPerOpSettings();

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

@@ -100,7 +100,7 @@ public class Client implements AutoCloseable {
   private final BigtableTimer sessionTimer;
 
   private final CallOptions defaultCallOptions;
-  private final ChannelPool channelPool;
+  private final Resource<ChannelPool> channelPool;
   private final Resource<Metrics> metrics;
   private final Resource<ClientConfigurationManager> configManager;
 
@@ -185,22 +185,26 @@ 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.<Executor>createOwned(
-            userCallbackExecutor, () -> shutdownAndAwait(userCallbackExecutor)));
+            userCallbackExecutor, () -> shutdownAndAwait(userCallbackExecutor)),
+        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<Executor> userCallbackExecutor)
+      Resource<Executor> userCallbackExecutor,
+      ChannelProvider channelProvider)
       throws IOException {
     this.featureFlags = featureFlags;
     this.clientInfo = clientInfo;
@@ -224,13 +228,38 @@ 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 and other resources should be created with
+   * Resource.createShared() and closed when the factory closes.
+   */
+  public Client(
+      FeatureFlags featureFlags,
+      ClientInfo clientInfo,
+      Resource<Metrics> metrics,
+      Resource<ClientConfigurationManager> configManager,
+      Resource<ScheduledExecutorService> bgExecutor,
+      Resource<Executor> userCallbackExecutor,
+      Resource<ChannelPool> sharedChannelPool) {
+    this.featureFlags = featureFlags;
+    this.clientInfo = clientInfo;
+    this.metrics = metrics;
+    this.configManager = configManager;
+    this.backgroundExecutor = bgExecutor;
+    this.userCallbackExecutor = userCallbackExecutor;
+    this.channelPool = sharedChannelPool;
+    this.sessionTimer = new HashedWheelTimer("bigtable-session-timer");
+    defaultCallOptions = CallOptions.DEFAULT;
   }
 
   @Override
@@ -329,7 +358,7 @@ public TableAsync openTableAsync(String tableId, Permission permission) {
               featureFlags,
               clientInfo,
               configManager.get(),
-              channelPool,
+              channelPool.get(),
               defaultCallOptions,
               tableId,
               permission,
@@ -353,7 +382,7 @@ public AuthorizedViewAsync openAuthorizedViewAsync(
               featureFlags,
               clientInfo,
               configManager.get(),
-              channelPool,
+              channelPool.get(),
               defaultCallOptions,
               tableId,
               viewId,
@@ -378,7 +407,7 @@ public MaterializedViewAsync openMaterializedViewAsync(
               featureFlags,
               clientInfo,
               configManager.get(),
-              channelPool,
+              channelPool.get(),
               defaultCallOptions,
               viewId,
               permission,
@@ -391,6 +420,21 @@ public MaterializedViewAsync openMaterializedViewAsync(
     }
   }
 
+  /** 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;
+  }
+
+  /** Returns the underlying user callback executor (e.g. for sharing with factory children). */
+  public Executor getUserCallbackExecutor() {
+    return userCallbackExecutor.get();
+  }
+
   public static class Resource<T> {
     private final T value;
     private final 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;
@@ -66,14 +68,20 @@ static <ReqT extends Message> TableBase createAndStart(
       Executor backgroundExecutor,
       Executor userCallbackExecutor) {
 
+    // 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,
             timer,