fix(spanner): fix grpc-gcp affinity cleanup and multiplexed channel usage leaks (#12726)

## Summary

This change fixes two regressions in the Spanner Java client introduced
around multiplexed session initialization and
  grpc-gcp affinity handling.

### 1. Fix static `CHANNEL_USAGE` leak in
`MultiplexedSessionDatabaseClient`

  Fixes: #12693
`MultiplexedSessionDatabaseClient` stored per-`SpannerImpl` channel
usage state in a static map but never removed
entries on close. After multiplexed client creation became unconditional
in `SpannerImpl.getDatabaseClient()`,
applications that repeatedly created and closed `Spanner` instances
could retain closed `SpannerImpl` objects, gRPC
  channels, and related transport state indefinitely.

  This change:
- replaces the static `Map<SpannerImpl, BitSet>` with reference-counted
shared state
- removes the map entry when the last `MultiplexedSessionDatabaseClient`
for a given `SpannerImpl` closes
- preserves sharing semantics for multiple database clients created from
the same `SpannerImpl`

### 2. Stop using bitset / bounded `% numChannels` affinity for grpc-gcp

For grpc-gcp-enabled paths, channel affinity should use the raw random
channel hint and rely on explicit unbind /
  cleanup, rather than:
  - bitset reservation
  - collapsing the hint with `% numChannels`

  This change:
  - keeps grpc-gcp on raw random affinity keys in `GapicSpannerRpc`
  - removes `% numChannels` mapping from the grpc-gcp call path
  - keeps the old non-grpc-gcp GAX affinity behavior unchanged

### 3. Add explicit grpc-gcp affinity cleanup for multi-use read-only
transaction close

Multi-use read-only transactions reuse a single random channel hint for
the lifetime of the transaction. That hint
should remain stable across all reads in the transaction, then be
explicitly cleaned up when the transaction closes.

  This change:
- adds a new RPC cleanup hook that carries both transaction id and
channel hint
  - invokes that cleanup from multi-use read-only transaction `close()`
- unbinds grpc-gcp affinity on transaction close without issuing an
`ExecuteSql` RPC to Spanner
  - handles both:
    - location API disabled: cleanup via the underlying grpc-gcp channel
- location API enabled: cleanup via `KeyAwareChannel`, using the routed
endpoint associated with the transaction

### 4. Apply grpc-gcp affinity cleanup settings without implicitly
enabling DCP

grpc-gcp affinity cleanup settings (`affinityKeyLifetime`,
`cleanupInterval`) should be applied whether or not dynamic channel pool
(DCP) is enabled.

  This change:
  - always passes grpc-gcp channel-pool options when grpc-gcp is enabled
  - preserves full pool settings only when DCP is enabled
  - passes cleanup-only channel-pool options when DCP is disabled:
    - `affinityKeyLifetime`
    - `cleanupInterval`
    - dynamic scaling explicitly disabled

Author: rahul2393

java-spanner/.gitignore

@@ -466,7 +466,7 @@ ByteString getTransactionId() {
     public void close() {
       ByteString id = getTransactionId();
       if (id != null && !id.isEmpty()) {
-        rpc.clearTransactionAffinity(id);
+        rpc.clearTransactionAndChannelAffinity(id, Option.CHANNEL_HINT.getLong(channelHint));
       }
       super.close();
     }

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/AbstractReadContext.java

@@ -161,7 +161,17 @@ public void close() {
    * Keeps track of which channels have been 'given' to single-use transactions for a given Spanner
    * instance.
    */
-  private static final Map<SpannerImpl, BitSet> CHANNEL_USAGE = new HashMap<>();
+  private static final class SharedChannelUsage {
+    private final BitSet channelUsage;
+    private int referenceCount;
+
+    private SharedChannelUsage(int numChannels) {
+      this.channelUsage = new BitSet(numChannels);
+      this.referenceCount = 0;
+    }
+  }
+
+  private static final Map<SpannerImpl, SharedChannelUsage> CHANNEL_USAGE = new HashMap<>();
 
   private static final EnumSet<ErrorCode> RETRYABLE_ERROR_CODES =
       EnumSet.of(ErrorCode.DEADLINE_EXCEEDED, ErrorCode.RESOURCE_EXHAUSTED, ErrorCode.UNAVAILABLE);
@@ -182,6 +192,8 @@ public void close() {
 
   private final SessionClient sessionClient;
 
+  private final SpannerImpl spanner;
+
   private final TraceWrapper tracer;
 
   /** The current multiplexed session that is used by this client. */
@@ -213,15 +225,20 @@ public void close() {
 
   @VisibleForTesting
   MultiplexedSessionDatabaseClient(SessionClient sessionClient, Clock clock) {
-    this.numChannels = sessionClient.getSpanner().getOptions().getNumChannels();
+    this.spanner = sessionClient.getSpanner();
+    this.numChannels = spanner.getOptions().getNumChannels();
     synchronized (CHANNEL_USAGE) {
-      CHANNEL_USAGE.putIfAbsent(sessionClient.getSpanner(), new BitSet(numChannels));
-      this.channelUsage = CHANNEL_USAGE.get(sessionClient.getSpanner());
+      SharedChannelUsage sharedChannelUsage = CHANNEL_USAGE.get(this.spanner);
+      if (sharedChannelUsage == null) {
+        sharedChannelUsage = new SharedChannelUsage(numChannels);
+        CHANNEL_USAGE.put(this.spanner, sharedChannelUsage);
+      }
+      sharedChannelUsage.referenceCount++;
+      this.channelUsage = sharedChannelUsage.channelUsage;
     }
     this.sessionExpirationDuration =
         Duration.ofMillis(
-            sessionClient
-                .getSpanner()
+            spanner
                 .getOptions()
                 .getSessionPoolOptions()
                 .getMultiplexedSessionMaintenanceDuration()
@@ -354,10 +371,23 @@ AtomicLong getNumSessionsReleased() {
   }
 
   void close() {
+    boolean releaseChannelUsage = false;
     synchronized (this) {
       if (!this.isClosed) {
         this.isClosed = true;
         this.maintainer.stop();
+        releaseChannelUsage = true;
+      }
+    }
+    if (releaseChannelUsage) {
+      synchronized (CHANNEL_USAGE) {
+        SharedChannelUsage sharedChannelUsage = CHANNEL_USAGE.get(this.spanner);
+        if (sharedChannelUsage != null) {
+          sharedChannelUsage.referenceCount--;
+          if (sharedChannelUsage.referenceCount == 0) {
+            CHANNEL_USAGE.remove(this.spanner);
+          }
+        }
       }
     }
   }

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/MultiplexedSessionDatabaseClient.java

@@ -64,6 +64,7 @@
 import com.google.cloud.grpc.GcpManagedChannel;
 import com.google.cloud.grpc.GcpManagedChannelBuilder;
 import com.google.cloud.grpc.GcpManagedChannelOptions;
+import com.google.cloud.grpc.GcpManagedChannelOptions.GcpChannelPoolOptions;
 import com.google.cloud.grpc.GcpManagedChannelOptions.GcpMetricsOptions;
 import com.google.cloud.grpc.GrpcTransportOptions;
 import com.google.cloud.grpc.fallback.GcpFallbackChannel;
@@ -301,6 +302,7 @@ public class GapicSpannerRpc implements SpannerRpc {
   private final boolean isGrpcGcpExtensionEnabled;
   private final boolean isDynamicChannelPoolEnabled;
   @Nullable private final KeyAwareChannel keyAwareChannel;
+  @Nullable private final GcpManagedChannel grpcGcpChannel;
 
   private final GrpcCallContext baseGrpcCallContext;
 
@@ -420,6 +422,7 @@ public GapicSpannerRpc(final SpannerOptions options) {
                 .build();
         ClientContext clientContext = ClientContext.create(spannerStubSettings);
         this.keyAwareChannel = extractKeyAwareChannel(clientContext.getTransportChannel());
+        this.grpcGcpChannel = extractGrpcGcpChannel(clientContext.getTransportChannel());
         this.spannerStub =
             GrpcSpannerStubWithStubSettingsAndClientContext.create(
                 spannerStubSettings, clientContext);
@@ -540,6 +543,7 @@ public <RequestT, ResponseT> UnaryCallable<RequestT, ResponseT> createUnaryCalla
       }
     } else {
       this.keyAwareChannel = null;
+      this.grpcGcpChannel = null;
       this.databaseAdminStub = null;
       this.instanceAdminStub = null;
       this.spannerStub = null;
@@ -589,13 +593,35 @@ private static KeyAwareChannel extractKeyAwareChannel(TransportChannel transport
     return null;
   }
 
+  @Nullable
+  private static GcpManagedChannel extractGrpcGcpChannel(TransportChannel transportChannel) {
+    if (!(transportChannel instanceof GrpcTransportChannel)) {
+      return null;
+    }
+    Channel channel = ((GrpcTransportChannel) transportChannel).getChannel();
+    if (channel instanceof GcpManagedChannel) {
+      return (GcpManagedChannel) channel;
+    }
+    return null;
+  }
+
   @Override
   public void clearTransactionAffinity(ByteString transactionId) {
     if (keyAwareChannel != null) {
       keyAwareChannel.clearTransactionAffinity(transactionId);
     }
   }
 
+  @Override
+  public void clearTransactionAndChannelAffinity(
+      ByteString transactionId, @Nullable Long channelHint) {
+    if (keyAwareChannel != null) {
+      keyAwareChannel.clearTransactionAndChannelAffinity(transactionId, channelHint);
+      return;
+    }
+    GrpcGcpAffinityUtil.clearChannelHintAffinity(grpcGcpChannel, channelHint);
+  }
+
   private static String parseGrpcGcpApiConfig() {
     try {
       return Resources.toString(
@@ -772,16 +798,35 @@ private static GcpManagedChannelOptions grpcGcpOptionsWithMetricsAndDcp(SpannerO
     }
     optionsBuilder.withMetricsOptions(metricsOptionsBuilder.build());
 
-    // Configure dynamic channel pool options if enabled.
-    // Uses the GcpChannelPoolOptions from SpannerOptions, which contains Spanner-specific defaults
-    // or user-provided configuration.
-    if (options.isDynamicChannelPoolEnabled()) {
-      optionsBuilder.withChannelPoolOptions(options.getGcpChannelPoolOptions());
+    // Always pass channel-pool options when grpc-gcp is enabled so affinity cleanup settings are
+    // applied regardless of whether dynamic channel pool is enabled. In the non-DCP path, only
+    // propagate the affinity cleanup configuration to avoid implicitly turning on dynamic scaling.
+    if (options.isGrpcGcpExtensionEnabled()) {
+      optionsBuilder.withChannelPoolOptions(getGrpcGcpChannelPoolOptions(options));
     }
 
     return optionsBuilder.build();
   }
 
+  @VisibleForTesting
+  static GcpChannelPoolOptions getGrpcGcpChannelPoolOptions(SpannerOptions options) {
+    GcpChannelPoolOptions channelPoolOptions = options.getGcpChannelPoolOptions();
+    if (options.isDynamicChannelPoolEnabled()) {
+      return channelPoolOptions;
+    }
+
+    // When DCP is disabled, Spanner's numChannels should still produce a fixed grpc-gcp channel
+    // pool instead of only capping the pool's maximum size.
+    return GcpChannelPoolOptions.newBuilder()
+        .setMaxSize(options.getNumChannels())
+        .setMinSize(options.getNumChannels())
+        .setInitSize(options.getNumChannels())
+        .disableDynamicScaling()
+        .setAffinityKeyLifetime(channelPoolOptions.getAffinityKeyLifetime())
+        .setCleanupInterval(channelPoolOptions.getCleanupInterval())
+        .build();
+  }
+
   @SuppressWarnings("rawtypes")
   private static void maybeEnableGrpcGcpExtension(
       InstantiatingGrpcChannelProvider.Builder defaultChannelProviderBuilder,
@@ -793,21 +838,19 @@ private static void maybeEnableGrpcGcpExtension(
     final String jsonApiConfig = parseGrpcGcpApiConfig();
     final GcpManagedChannelOptions grpcGcpOptions = grpcGcpOptionsWithMetricsAndDcp(options);
 
-    // When dynamic channel pool is enabled, use the DCP initial size as the pool size.
-    // When disabled, use the explicitly configured numChannels.
-    final int poolSize = options.isDynamicChannelPoolEnabled() ? 0 : options.getNumChannels();
-
     ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder> baseConfigurator =
         defaultChannelProviderBuilder.getChannelConfigurator();
     ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder> apiFunction =
         channelBuilder -> {
           if (baseConfigurator != null) {
             channelBuilder = baseConfigurator.apply(channelBuilder);
           }
+          // The grpc-gcp pool is configured entirely through GcpChannelPoolOptions above. Avoid
+          // the deprecated setPoolSize path, which only adjusts maxSize and does not eagerly create
+          // a fixed-size pool.
           return GcpManagedChannelBuilder.forDelegateBuilder(channelBuilder)
               .withApiConfigJsonString(jsonApiConfig)
-              .withOptions(grpcGcpOptions)
-              .setPoolSize(poolSize);
+              .withOptions(grpcGcpOptions);
         };
 
     // Disable the GAX channel pooling functionality by setting the GAX channel pool size to 1.
@@ -2275,20 +2318,9 @@ <ReqT, RespT> GrpcCallContext newCallContext(
     Long affinity = options == null ? null : Option.CHANNEL_HINT.getLong(options);
     if (affinity != null) {
       if (this.isGrpcGcpExtensionEnabled) {
-        // Set channel affinity in gRPC-GCP.
-        String affinityKey;
-        if (this.isDynamicChannelPoolEnabled) {
-          // When dynamic channel pooling is enabled, we use the raw affinity value as the key.
-          // This allows grpc-gcp to use round-robin for new keys, enabling new channels
-          // (created during scale-up) to receive requests. The affinity key lifetime setting
-          // ensures the affinity map doesn't grow unbounded.
-          affinityKey = String.valueOf(affinity);
-        } else {
-          // When DCP is disabled, compute bounded channel hint to prevent
-          // gRPC-GCP affinity map from getting unbounded.
-          int boundedChannelHint = affinity.intValue() % this.numChannels;
-          affinityKey = String.valueOf(boundedChannelHint);
-        }
+        // Set channel affinity in gRPC-GCP. Always use the raw affinity value as the key.
+        // Cleanup is handled explicitly by unbind on terminal/single-use operations.
+        String affinityKey = String.valueOf(affinity);
         context =
             context.withCallOptions(
                 context.getCallOptions().withOption(GcpManagedChannel.AFFINITY_KEY, affinityKey));

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner.spi.v1;
+
+import com.google.cloud.grpc.GcpManagedChannel;
+import com.google.spanner.v1.ExecuteSqlRequest;
+import com.google.spanner.v1.ResultSet;
+import com.google.spanner.v1.SpannerGrpc;
+import io.grpc.CallOptions;
+import io.grpc.ClientCall;
+import io.grpc.ManagedChannel;
+import javax.annotation.Nullable;
+
+final class GrpcGcpAffinityUtil {
+  private GrpcGcpAffinityUtil() {}
+
+  static void clearChannelHintAffinity(
+      @Nullable ManagedChannel channel, @Nullable Long channelHint) {
+    if (!(channel instanceof GcpManagedChannel) || channelHint == null) {
+      return;
+    }
+    // TODO: Replace this synthetic call once grpc-gcp exposes a direct API for unbinding
+    // affinity keys without creating and immediately cancelling a ClientCall.
+    ClientCall<ExecuteSqlRequest, ResultSet> call =
+        channel.newCall(
+            SpannerGrpc.getExecuteSqlMethod(),
+            CallOptions.DEFAULT
+                .withOption(GcpManagedChannel.AFFINITY_KEY, String.valueOf(channelHint))
+                .withOption(GcpManagedChannel.UNBIND_AFFINITY_KEY, true));
+    call.cancel("Cloud Spanner transaction closed", null);
+  }
+}

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GrpcGcpAffinityUtil.java

@@ -304,6 +304,21 @@ void clearTransactionAffinity(ByteString transactionId) {
     clearAffinity(transactionId);
   }
 
+  void clearTransactionAndChannelAffinity(ByteString transactionId, @Nullable Long channelHint) {
+    String address = transactionAffinities.remove(transactionId);
+    readOnlyTxPreferLeader.invalidate(transactionId);
+    if (channelHint != null) {
+      ManagedChannel channel = defaultChannel;
+      if (address != null) {
+        ChannelEndpoint endpoint = endpointCache.getIfPresent(address);
+        if (endpoint != null) {
+          channel = endpoint.getChannel();
+        }
+      }
+      GrpcGcpAffinityUtil.clearChannelHintAffinity(channel, channelHint);
+    }
+  }
+
   private void maybeExcludeEndpointOnNextCall(
       @Nullable ChannelEndpoint endpoint, @Nullable String logicalRequestKey) {
     if (endpoint == null || logicalRequestKey == null) {

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/KeyAwareChannel.java

@@ -199,6 +199,15 @@ default RequestIdCreator getRequestIdCreator() {
   /** Clears any client-side affinity associated with the given transaction id. */
   default void clearTransactionAffinity(ByteString transactionId) {}
 
+  /**
+   * Clears any client-side transaction affinity and transport-level channel affinity associated
+   * with the given transaction.
+   */
+  default void clearTransactionAndChannelAffinity(
+      ByteString transactionId, @Nullable Long channelHint) {
+    clearTransactionAffinity(transactionId);
+  }
+
   // Instance admin APIs.
   Paginated<InstanceConfig> listInstanceConfigs(int pageSize, @Nullable String pageToken)
       throws SpannerException;