fix(spanner): don't set sticky AFFINITY_KEY for multiplexed sessions

Under grpc-gcp (default since 6.105.0, #4239), newCallContext set
GcpManagedChannel.AFFINITY_KEY on every data RPC. GcpManagedChannel
binds each new key via pickLeastBusyChannel, which reads
activeStreamsCount before any concurrent caller's start() has
incremented it (tiebreak channelRefs[0]). A high-concurrency cold start
therefore binds keys to channel 0 and the bindings are sticky, so RPCs
funnel through one HTTP/2 connection and queue at MAX_CONCURRENT_STREAMS.

The static-numChannels path bounded the key to ~2*numChannels-1 distinct
values, making the collapse permanent (~6x throughput regression at 400
concurrent). The dynamic-channel-pool path used per-transaction random
keys and largely self-corrected, but a few BitSet-recycled hints still
sticky-bound, leaving a p99 tail.

Multiplexed sessions get no backend-locality benefit from sticky
per-transaction channel affinity, so don't set the key under grpc-gcp at
all. getChannelRef(null) does a fresh per-call least-busy pick with no
sticky binding and no affinity-map growth.

Drops the now-unobservable distinct-AFFINITY_KEY assertions from
RetryOnDifferentGrpcChannelMockServerTest; the request-count and session
assertions still cover the retry loop.

Author: akash329d

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

@@ -2275,30 +2275,12 @@ <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);
-        }
-        context =
-            context.withCallOptions(
-                context.getCallOptions().withOption(GcpManagedChannel.AFFINITY_KEY, affinityKey));
-        // Check if the caller wants to unbind the affinity key after this call completes.
-        Boolean unbind = Option.UNBIND_CHANNEL_HINT.get(options);
-        if (Boolean.TRUE.equals(unbind)) {
-          context =
-              context.withCallOptions(
-                  context.getCallOptions().withOption(GcpManagedChannel.UNBIND_AFFINITY_KEY, true));
-        }
+        // Do not set AFFINITY_KEY under grpc-gcp. Multiplexed sessions get no backend-locality
+        // benefit from sticky per-transaction channel affinity, and setting one causes
+        // GcpManagedChannel.pickLeastBusyChannel to permanently bind keys to channel 0 under a
+        // concurrent warmup burst (it reads activeStreamsCount before any caller's start() has
+        // incremented it). With no key, getChannelRef(null) does a fresh per-call least-busy pick
+        // with no sticky binding and no affinity-map growth.
       } else {
         // Set channel affinity in GAX.
         context = context.withChannelAffinity(affinity.intValue());

java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/RetryOnDifferentGrpcChannelMockServerTest.java

@@ -23,30 +23,18 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assume.assumeFalse;
 
-import com.google.api.gax.grpc.GrpcInterceptorProvider;
 import com.google.cloud.NoCredentials;
-import com.google.cloud.grpc.GcpManagedChannel;
 import com.google.cloud.spanner.MockSpannerServiceImpl.SimulatedExecutionTime;
 import com.google.cloud.spanner.connection.AbstractMockServerTest;
-import com.google.common.collect.ImmutableList;
 import com.google.spanner.v1.BatchCreateSessionsRequest;
 import com.google.spanner.v1.BeginTransactionRequest;
 import com.google.spanner.v1.ExecuteSqlRequest;
-import com.google.spanner.v1.SpannerGrpc;
-import io.grpc.CallOptions;
-import io.grpc.Channel;
-import io.grpc.ClientCall;
-import io.grpc.ClientInterceptor;
 import io.grpc.Context;
 import io.grpc.Deadline;
 import io.grpc.ManagedChannelBuilder;
-import io.grpc.MethodDescriptor;
 import io.grpc.Status;
 import java.time.Duration;
-import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -62,11 +50,6 @@
 
 @RunWith(JUnit4.class)
 public class RetryOnDifferentGrpcChannelMockServerTest extends AbstractMockServerTest {
-  /** Tracks the logical affinity keys before grpc-gcp routes the request. */
-  private static final Map<String, Set<String>> LOGICAL_AFFINITY_KEYS = new HashMap<>();
-
-  /** Tracks how many calls explicitly request grpc-gcp affinity unbind. */
-  private static final Map<String, Integer> UNBIND_AFFINITY_CALL_COUNTS = new HashMap<>();
 
   @BeforeClass
   public static void setupAndStartServer() throws Exception {
@@ -82,52 +65,16 @@ public static void removeSystemProperty() {
 
   @After
   public void clearRequests() {
-    LOGICAL_AFFINITY_KEYS.clear();
-    UNBIND_AFFINITY_CALL_COUNTS.clear();
     mockSpanner.clearRequests();
     mockSpanner.removeAllExecutionTimes();
   }
 
-  /**
-   * Creates a client interceptor that captures the logical affinity key before grpc-gcp routes the
-   * request. This allows us to verify that retry logic uses distinct logical channel hints, even
-   * when DCP maps them to fewer physical channels.
-   */
-  static GrpcInterceptorProvider createAffinityKeyInterceptorProvider() {
-    return () ->
-        ImmutableList.of(
-            new ClientInterceptor() {
-              @Override
-              public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
-                  MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
-                String methodName = method.getFullMethodName();
-                // Capture the AFFINITY_KEY before grpc-gcp processes it.
-                String affinityKey = callOptions.getOption(GcpManagedChannel.AFFINITY_KEY);
-                if (affinityKey != null) {
-                  synchronized (LOGICAL_AFFINITY_KEYS) {
-                    Set<String> keys =
-                        LOGICAL_AFFINITY_KEYS.computeIfAbsent(methodName, k -> new HashSet<>());
-                    keys.add(affinityKey);
-                  }
-                }
-                if (Boolean.TRUE.equals(
-                    callOptions.getOption(GcpManagedChannel.UNBIND_AFFINITY_KEY))) {
-                  synchronized (UNBIND_AFFINITY_CALL_COUNTS) {
-                    UNBIND_AFFINITY_CALL_COUNTS.merge(methodName, 1, Integer::sum);
-                  }
-                }
-                return next.newCall(method, callOptions);
-              }
-            });
-  }
-
   SpannerOptions.Builder createSpannerOptionsBuilder() {
     return SpannerOptions.newBuilder()
         .setProjectId("my-project")
         .setHost(String.format("http://localhost:%d", getPort()))
         .setChannelConfigurator(ManagedChannelBuilder::usePlaintext)
-        .setCredentials(NoCredentials.getInstance())
-        .setInterceptorProvider(createAffinityKeyInterceptorProvider());
+        .setCredentials(NoCredentials.getInstance());
   }
 
   @Test
@@ -163,12 +110,6 @@ public void testReadWriteTransaction_retriesOnNewChannel() {
     List<BeginTransactionRequest> requests =
         mockSpanner.getRequestsOfType(BeginTransactionRequest.class);
     assertNotEquals(requests.get(0).getSession(), requests.get(1).getSession());
-    // Verify that the retry used 2 distinct logical affinity keys (before grpc-gcp routing).
-    assertEquals(
-        2,
-        LOGICAL_AFFINITY_KEYS
-            .getOrDefault("google.spanner.v1.Spanner/BeginTransaction", new HashSet<>())
-            .size());
   }
 
   @Test
@@ -208,13 +149,6 @@ public void testReadWriteTransaction_stopsRetrying() {
       Set<String> sessions =
           requests.stream().map(BeginTransactionRequest::getSession).collect(Collectors.toSet());
       assertEquals(numChannels, sessions.size());
-      // Verify that the retry logic used distinct logical affinity keys (before grpc-gcp routing).
-      // This confirms each retry attempt targeted a different logical channel.
-      assertEquals(
-          numChannels,
-          LOGICAL_AFFINITY_KEYS
-              .getOrDefault("google.spanner.v1.Spanner/BeginTransaction", new HashSet<>())
-              .size());
     }
   }
 
@@ -284,13 +218,6 @@ public void testDenyListedChannelIsCleared() {
       // of the first transaction. That fails, the session is deny-listed, the transaction is
       // retried on yet another session and succeeds.
       assertEquals(numChannels + 1, sessions.size());
-      // Verify that the retry logic used distinct logical affinity keys (before grpc-gcp routing).
-      // This confirms each retry attempt targeted a different logical channel.
-      assertEquals(
-          numChannels,
-          LOGICAL_AFFINITY_KEYS
-              .getOrDefault("google.spanner.v1.Spanner/BeginTransaction", new HashSet<>())
-              .size());
       assertEquals(numChannels, mockSpanner.countRequestsOfType(BatchCreateSessionsRequest.class));
     }
   }
@@ -316,17 +243,6 @@ public void testSingleUseQuery_retriesOnNewChannel() {
     List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
     // The requests use the same multiplexed session.
     assertEquals(requests.get(0).getSession(), requests.get(1).getSession());
-    // Verify that the retry used 2 distinct logical affinity keys (before grpc-gcp routing).
-    assertEquals(
-        2,
-        LOGICAL_AFFINITY_KEYS
-            .getOrDefault("google.spanner.v1.Spanner/ExecuteStreamingSql", new HashSet<>())
-            .size());
-    assertEquals(
-        2,
-        UNBIND_AFFINITY_CALL_COUNTS
-            .getOrDefault(SpannerGrpc.getExecuteStreamingSqlMethod().getFullMethodName(), 0)
-            .intValue());
   }
 
   @Test
@@ -354,17 +270,6 @@ public void testSingleUseQuery_stopsRetrying() {
       // Verify that the retry mechanism is working (made numChannels requests).
       int totalRequests = mockSpanner.countRequestsOfType(ExecuteSqlRequest.class);
       assertEquals(numChannels, totalRequests);
-      // Verify each attempt used a distinct logical affinity key (before grpc-gcp routing).
-      int distinctLogicalKeys =
-          LOGICAL_AFFINITY_KEYS
-              .getOrDefault("google.spanner.v1.Spanner/ExecuteStreamingSql", new HashSet<>())
-              .size();
-      assertEquals(totalRequests, distinctLogicalKeys);
-      assertEquals(
-          totalRequests,
-          UNBIND_AFFINITY_CALL_COUNTS
-              .getOrDefault(SpannerGrpc.getExecuteStreamingSqlMethod().getFullMethodName(), 0)
-              .intValue());
     }
   }