feat: add KeyAwareChannel and location-aware routing integration

This commit adds the KeyAwareChannel class and integrates all components
for location-aware routing in the Spanner client.

Key components:
- ChannelFinder: Orchestrates routing decisions using caches
- KeyAwareChannel: Custom ManagedChannel that intercepts key-aware methods
- SpannerOptions integration: Enables feature via setExperimentalHost()
- GapicSpannerRpc modifications: Wire in KeyAwareChannel

Activation requires:
- SpannerOptions.Builder.setExperimentalHost() with location-aware endpoint
- GOOGLE_SPANNER_EXPERIMENTAL_LOCATION_API=true environment variable

This is part of the experimental location-aware routing for improved latency.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: rahul2393

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

@@ -257,6 +257,7 @@ public static GcpChannelPoolOptions createDefaultDynamicChannelPoolOptions() {
   private final boolean enableEndToEndTracing;
   private final String monitoringHost;
   private final TransactionOptions defaultTransactionOptions;
+  private final boolean isExperimentalHost;
 
   enum TracingFramework {
     OPEN_CENSUS,
@@ -922,6 +923,12 @@ protected SpannerOptions(Builder builder) {
     enableEndToEndTracing = builder.enableEndToEndTracing;
     monitoringHost = builder.monitoringHost;
     defaultTransactionOptions = builder.defaultTransactionOptions;
+    isExperimentalHost = builder.isExperimentalHost;
+  }
+
+  /** Returns true if an experimental host is configured. */
+  public boolean isExperimentalHost() {
+    return isExperimentalHost;
   }
 
   private String getResolvedUniverseDomain() {
@@ -987,6 +994,15 @@ default String getMonitoringHost() {
     default GoogleCredentials getDefaultExperimentalHostCredentials() {
       return null;
     }
+
+    /**
+     * Returns true if the experimental location API (SpanFE bypass) should be enabled. When
+     * enabled, the client will use location-aware routing to send requests directly to the
+     * appropriate Spanner server.
+     */
+    default boolean isEnableLocationApi() {
+      return false;
+    }
   }
 
   static final String DEFAULT_SPANNER_EXPERIMENTAL_HOST_CREDENTIALS =
@@ -1011,6 +1027,8 @@ private static class SpannerEnvironmentImpl implements SpannerEnvironment {
     private static final String SPANNER_DISABLE_DIRECT_ACCESS_GRPC_BUILTIN_METRICS =
         "SPANNER_DISABLE_DIRECT_ACCESS_GRPC_BUILTIN_METRICS";
     private static final String SPANNER_MONITORING_HOST = "SPANNER_MONITORING_HOST";
+    private static final String GOOGLE_SPANNER_EXPERIMENTAL_LOCATION_API =
+        "GOOGLE_SPANNER_EXPERIMENTAL_LOCATION_API";
 
     private SpannerEnvironmentImpl() {}
 
@@ -1069,6 +1087,11 @@ public String getMonitoringHost() {
     public GoogleCredentials getDefaultExperimentalHostCredentials() {
       return getOAuthTokenFromFile(System.getenv(DEFAULT_SPANNER_EXPERIMENTAL_HOST_CREDENTIALS));
     }
+
+    @Override
+    public boolean isEnableLocationApi() {
+      return Boolean.parseBoolean(System.getenv(GOOGLE_SPANNER_EXPERIMENTAL_LOCATION_API));
+    }
   }
 
   /** Builder for {@link SpannerOptions} instances. */
@@ -1141,6 +1164,7 @@ public static class Builder
     private SslContext mTLSContext = null;
     private String experimentalHost = null;
     private boolean usePlainText = false;
+    private boolean isExperimentalHost = false;
     private TransactionOptions defaultTransactionOptions = TransactionOptions.getDefaultInstance();
 
     private static String createCustomClientLibToken(String token) {
@@ -1689,6 +1713,7 @@ public Builder setExperimentalHost(String host) {
       super.setProjectId(EXPERIMENTAL_HOST_PROJECT_ID);
       setSessionPoolOption(SessionPoolOptions.newBuilder().setExperimentalHost().build());
       this.experimentalHost = host;
+      this.isExperimentalHost = true;
       return this;
     }
 

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

@@ -0,0 +1,119 @@
+/*
+ * 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.spanner.v1.CacheUpdate;
+import com.google.spanner.v1.ReadRequest;
+import com.google.spanner.v1.RoutingHint;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import javax.annotation.Nullable;
+
+/**
+ * ChannelFinder is responsible for finding the correct Spanner server to route RPCs to.
+ *
+ * <p>It uses a {@link KeyRecipeCache} and a {@link KeyRangeCache} to store metadata about the
+ * database, including key recipes and range information. This metadata is updated through the
+ * {@link #update(CacheUpdate)} method.
+ *
+ * <p>The {@link #findServer(ReadRequest.Builder)} method is used to determine the appropriate
+ * server for a given read request.
+ */
+public final class ChannelFinder {
+  private final String deployment;
+  private final String databaseUri;
+  private final KeyRangeCache rangeCache;
+  private final KeyRecipeCache recipeCache;
+  private final ReadWriteLock lock = new ReentrantReadWriteLock();
+  private long databaseId = 0L;
+  private final ChannelFinderServerFactory serverFactory;
+
+  public ChannelFinder(
+      ChannelFinderServerFactory serverFactory, String deployment, String databaseUri) {
+    this.serverFactory = serverFactory;
+    this.deployment = deployment;
+    this.databaseUri = databaseUri;
+    this.rangeCache = new KeyRangeCache(serverFactory);
+    this.recipeCache = new KeyRecipeCache();
+  }
+
+  /**
+   * Updates the cache with new metadata.
+   *
+   * @param cacheUpdate The cache update information.
+   */
+  public void update(CacheUpdate cacheUpdate) {
+    lock.writeLock().lock();
+    try {
+      if (databaseId != cacheUpdate.getDatabaseId()) {
+        System.out.println("DEBUG [BYPASS]: Database ID changed from " + databaseId 
+            + " to " + cacheUpdate.getDatabaseId() + ", clearing caches");
+        recipeCache.clear();
+        rangeCache.clear();
+        databaseId = cacheUpdate.getDatabaseId();
+      }
+      recipeCache.addRecipes(cacheUpdate.getKeyRecipes());
+      rangeCache.addRanges(cacheUpdate);
+      System.out.println("DEBUG [BYPASS]: Cache updated. Current state:\n" + rangeCache.debugString());
+    } finally {
+      lock.writeLock().unlock();
+    }
+  }
+
+  /**
+   * Finds the server for a given ReadRequest.
+   *
+   * @param reqBuilder The ReadRequest builder.
+   * @return The server to route the request to, or null if an error occurs.
+   */
+  @Nullable
+  public ChannelFinderServer findServer(ReadRequest.Builder reqBuilder) {
+    RoutingHint.Builder hintBuilder = reqBuilder.getRoutingHintBuilder();
+    lock.readLock().lock();
+    try {
+      if (databaseId != 0) {
+        hintBuilder.setDatabaseId(databaseId);
+      }
+      System.out.println("DEBUG [BYPASS]: findServer - computing keys for table: " 
+          + reqBuilder.getTable());
+      recipeCache.computeKeys(reqBuilder); // Modifies hintBuilder within reqBuilder
+      System.out.println("DEBUG [BYPASS]: findServer - after computeKeys, key: " 
+          + hintBuilder.getKey().toStringUtf8());
+      ChannelFinderServer server = rangeCache.fillRoutingInfo(
+          reqBuilder.getSession(), false, hintBuilder);
+      System.out.println("DEBUG [BYPASS]: findServer - fillRoutingInfo returned server: " 
+          + (server != null ? server.getAddress() : "null"));
+      return server;
+    } finally {
+      lock.readLock().unlock();
+    }
+  }
+
+  /**
+   * Returns a debug string representation of the cache.
+   *
+   * @return A string containing debug information.
+   */
+  public String debugString() {
+    lock.readLock().lock();
+    try {
+      return rangeCache.debugString();
+    } finally {
+      lock.readLock().unlock();
+    }
+  }
+}

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

@@ -1,3 +1,19 @@
+/*
+ * 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 io.grpc.ManagedChannel;

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

@@ -1,3 +1,19 @@
+/*
+ * 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;
 
 public interface ChannelFinderServerFactory {

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

@@ -57,6 +57,7 @@
 import com.google.api.gax.rpc.UnavailableException;
 import com.google.api.gax.rpc.WatchdogProvider;
 import com.google.api.pathtemplate.PathTemplate;
+import com.google.auth.Credentials;
 import com.google.cloud.RetryHelper;
 import com.google.cloud.RetryHelper.RetryHelperException;
 import com.google.cloud.grpc.GcpManagedChannel;
@@ -209,12 +210,15 @@
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 import java.util.stream.Stream;
 import javax.annotation.Nullable;
 
@@ -223,6 +227,7 @@
 public class GapicSpannerRpc implements SpannerRpc {
   private static final PathTemplate PROJECT_NAME_TEMPLATE =
       PathTemplate.create("projects/{project}");
+  private static final Logger logger = Logger.getLogger(GapicSpannerRpc.class.getName());
   private static final PathTemplate OPERATION_NAME_TEMPLATE =
       PathTemplate.create("{database=projects/*/instances/*/databases/*}/operations/{operation}");
   private static final int MAX_MESSAGE_SIZE = 256 * 1024 * 1024;
@@ -285,6 +290,89 @@ public class GapicSpannerRpc implements SpannerRpc {
 
   private final GrpcCallContext baseGrpcCallContext;
 
+  private static class KeyAwareTransportChannelProvider implements TransportChannelProvider {
+    private final InstantiatingGrpcChannelProvider.Builder delegateBuilder;
+    private final TransportChannelProvider delegate;
+
+    public KeyAwareTransportChannelProvider(
+        InstantiatingGrpcChannelProvider.Builder delegateBuilder) {
+      this.delegateBuilder = delegateBuilder;
+      this.delegate = delegateBuilder.build();
+    }
+
+    @Override
+    public GrpcTransportChannel getTransportChannel() throws IOException {
+      return GrpcTransportChannel.newBuilder()
+          .setManagedChannel(KeyAwareChannel.create(delegateBuilder))
+          .build();
+    }
+
+    @Override
+    public String getTransportName() {
+      return delegate.getTransportName();
+    }
+
+    @Override
+    public boolean needsEndpoint() {
+      return delegate.needsEndpoint();
+    }
+
+    @Override
+    public boolean needsCredentials() {
+      return delegate.needsCredentials();
+    }
+
+    @Override
+    public boolean needsExecutor() {
+      return delegate.needsExecutor();
+    }
+
+    @Override
+    public boolean needsHeaders() {
+      return delegate.needsHeaders();
+    }
+
+    @Override
+    public boolean shouldAutoClose() {
+      return delegate.shouldAutoClose();
+    }
+
+    @Override
+    public TransportChannelProvider withEndpoint(String endpoint) {
+      return delegate.withEndpoint(endpoint);
+    }
+
+    @Override
+    public TransportChannelProvider withCredentials(Credentials credentials) {
+      return delegate.withCredentials(credentials);
+    }
+
+    @Override
+    public TransportChannelProvider withHeaders(java.util.Map<String, String> headers) {
+      return delegate.withHeaders(headers);
+    }
+
+    @Override
+    public TransportChannelProvider withPoolSize(int poolSize) {
+      return delegate.withPoolSize(poolSize);
+    }
+
+    @Override
+    public TransportChannelProvider withExecutor(ScheduledExecutorService executor) {
+      return delegate.withExecutor(executor);
+    }
+
+    @Override
+    public TransportChannelProvider withExecutor(Executor executor) {
+      return delegate.withExecutor(executor);
+    }
+
+    @Override
+    public boolean acceptsPoolSize() {
+      return delegate.acceptsPoolSize();
+    }
+  }
+
   public static GapicSpannerRpc create(SpannerOptions options) {
     return new GapicSpannerRpc(options);
   }
@@ -393,9 +481,33 @@ public GapicSpannerRpc(final SpannerOptions options) {
       // If it is enabled in options uses the channel pool provided by the gRPC-GCP extension.
       maybeEnableGrpcGcpExtension(defaultChannelProviderBuilder, options);
 
-      TransportChannelProvider channelProvider =
-          MoreObjects.firstNonNull(
-              options.getChannelProvider(), defaultChannelProviderBuilder.build());
+      TransportChannelProvider channelProvider;
+      // Enable KeyAwareChannel (SpanFE bypass / location API) only when BOTH conditions are met:
+      // 1. Using experimental host (setExperimentalHost was called)
+      // 2. GOOGLE_SPANNER_EXPERIMENTAL_LOCATION_API env var is set to "true"
+      // Default is DISABLED even for experimental host.
+      String locationApiEnvVar = System.getenv("GOOGLE_SPANNER_EXPERIMENTAL_LOCATION_API");
+      boolean isExperimentalHost = options.isExperimentalHost();
+      boolean envVarEnabled = "true".equalsIgnoreCase(locationApiEnvVar);
+
+      // Both conditions must be true to enable bypass
+      boolean enableLocationApi = isExperimentalHost && envVarEnabled;
+
+      logger.log(
+          Level.INFO,
+          "SpanFE bypass (KeyAwareChannel) configuration: "
+              + "GOOGLE_SPANNER_EXPERIMENTAL_LOCATION_API={0}, "
+              + "isExperimentalHost={1}, "
+              + "enableLocationApi={2}",
+          new Object[] {locationApiEnvVar, isExperimentalHost, enableLocationApi});
+
+      if (enableLocationApi) {
+        channelProvider = new KeyAwareTransportChannelProvider(defaultChannelProviderBuilder);
+      } else {
+        channelProvider =
+            MoreObjects.firstNonNull(
+                options.getChannelProvider(), defaultChannelProviderBuilder.build());
+      }
 
       CredentialsProvider credentialsProvider =
           GrpcTransportOptions.setUpCredentialsProvider(options);