Merge branch 'main' into renovate/com.google.cloud-sdk-platform-java-config-3.x

Author: rahul2393

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

@@ -0,0 +1,66 @@
+/*
+ * 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.api.core.InternalApi;
+import io.grpc.ManagedChannel;
+
+/**
+ * Represents a Spanner server endpoint for location-aware routing.
+ *
+ * <p>Each instance wraps a gRPC {@link ManagedChannel} connected to a specific Spanner server. The
+ * {@link ChannelEndpointCache} creates and caches these instances.
+ *
+ * <p>Implementations must be thread-safe as instances may be shared across multiple concurrent
+ * operations.
+ *
+ * @see ChannelEndpointCache
+ */
+@InternalApi
+public interface ChannelEndpoint {
+
+  /**
+   * Returns the network address of this server.
+   *
+   * @return the server address in "host:port" format
+   */
+  String getAddress();
+
+  /**
+   * Returns whether this server is ready to accept RPCs.
+   *
+   * <p>A server is considered unhealthy if:
+   *
+   * <ul>
+   *   <li>The underlying channel is shutdown or terminated
+   *   <li>The channel is in a transient failure state
+   * </ul>
+   *
+   * @return true if the server is healthy and ready to accept RPCs
+   */
+  boolean isHealthy();
+
+  /**
+   * Returns the gRPC channel for making RPCs to this server.
+   *
+   * <p>The returned channel is managed by the {@link ChannelEndpointCache} and should not be shut
+   * down directly by callers.
+   *
+   * @return the managed channel for this server
+   */
+  ManagedChannel getChannel();
+}

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

@@ -0,0 +1,79 @@
+/*
+ * 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.api.core.InternalApi;
+
+/**
+ * Cache for server connections used in location-aware routing.
+ *
+ * <p>Implementations are expected to cache {@link ChannelEndpoint} instances such that repeated
+ * calls with the same address return the same instance. This allows routing components to
+ * efficiently manage server references.
+ *
+ * <p>Implementations must be thread-safe. Multiple threads may concurrently call {@link
+ * #get(String)} with different addresses.
+ */
+@InternalApi
+public interface ChannelEndpointCache {
+
+  /**
+   * Returns the default channel endpoint.
+   *
+   * <p>The default channel is the original endpoint configured in {@link
+   * com.google.cloud.spanner.SpannerOptions}. It is used as a fallback when the location cache does
+   * not have routing information for a request.
+   *
+   * @return the default channel, never null
+   */
+  ChannelEndpoint defaultChannel();
+
+  /**
+   * Returns a cached server for the given address, creating it if needed.
+   *
+   * <p>If a server for this address already exists in the cache, the cached instance is returned.
+   * Otherwise, a new server connection is created and cached.
+   *
+   * @param address the server address in "host:port" format
+   * @return a server instance for the address, never null
+   * @throws com.google.cloud.spanner.SpannerException if the channel cannot be created
+   */
+  ChannelEndpoint get(String address);
+
+  /**
+   * Evicts a server from the cache and gracefully shuts down its channel.
+   *
+   * <p>This method should be called when a server becomes unhealthy or is no longer needed. The
+   * channel shutdown is graceful: existing RPCs are allowed to complete, but new RPCs will not be
+   * accepted on this channel.
+   *
+   * <p>If the address is not in the cache, this method does nothing.
+   *
+   * @param address the server address to evict
+   */
+  void evict(String address);
+
+  /**
+   * Shuts down all cached server connections.
+   *
+   * <p>This method should be called when the Spanner client is closed to release all resources.
+   * Each channel is shut down gracefully, allowing in-flight RPCs to complete.
+   *
+   * <p>After calling this method, the cache should not be used to create new connections.
+   */
+  void shutdown();
+}

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

@@ -0,0 +1,203 @@
+/*
+ * 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.api.core.InternalApi;
+import com.google.api.gax.grpc.GrpcTransportChannel;
+import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
+import com.google.api.gax.rpc.TransportChannelProvider;
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.common.annotations.VisibleForTesting;
+import io.grpc.ConnectivityState;
+import io.grpc.ManagedChannel;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * gRPC implementation of {@link ChannelEndpointCache}.
+ *
+ * <p>This cache creates and caches gRPC channels per address. It uses {@link
+ * InstantiatingGrpcChannelProvider#withEndpoint(String)} to create new channels with the same
+ * configuration but different endpoints, avoiding race conditions.
+ */
+@InternalApi
+class GrpcChannelEndpointCache implements ChannelEndpointCache {
+
+  /** Timeout for graceful channel shutdown. */
+  private static final long SHUTDOWN_TIMEOUT_SECONDS = 5;
+
+  private final InstantiatingGrpcChannelProvider baseProvider;
+  private final Map<String, GrpcChannelEndpoint> servers = new ConcurrentHashMap<>();
+  private final GrpcChannelEndpoint defaultEndpoint;
+  private final AtomicBoolean isShutdown = new AtomicBoolean(false);
+
+  /**
+   * Creates a new cache with the given channel provider.
+   *
+   * @param channelProvider the base provider used to create channels. New channels for different
+   *     endpoints are created using {@link InstantiatingGrpcChannelProvider#withEndpoint(String)}.
+   * @throws IOException if the default channel cannot be created
+   */
+  public GrpcChannelEndpointCache(InstantiatingGrpcChannelProvider channelProvider)
+      throws IOException {
+    this.baseProvider = channelProvider;
+    String defaultEndpoint = channelProvider.getEndpoint();
+    this.defaultEndpoint = new GrpcChannelEndpoint(defaultEndpoint, channelProvider);
+    this.servers.put(defaultEndpoint, this.defaultEndpoint);
+  }
+
+  @Override
+  public ChannelEndpoint defaultChannel() {
+    return defaultEndpoint;
+  }
+
+  @Override
+  public ChannelEndpoint get(String address) {
+    if (isShutdown.get()) {
+      throw SpannerExceptionFactory.newSpannerException(
+          ErrorCode.FAILED_PRECONDITION, "ChannelEndpointCache has been shut down");
+    }
+
+    return servers.computeIfAbsent(
+        address,
+        addr -> {
+          try {
+            // Create a new provider with the same config but different endpoint.
+            // This is thread-safe as withEndpoint() returns a new provider instance.
+            TransportChannelProvider newProvider = baseProvider.withEndpoint(addr);
+            return new GrpcChannelEndpoint(addr, newProvider);
+          } catch (IOException e) {
+            throw SpannerExceptionFactory.newSpannerException(
+                ErrorCode.INTERNAL, "Failed to create channel for address: " + addr, e);
+          }
+        });
+  }
+
+  @Override
+  public void evict(String address) {
+    if (defaultEndpoint.getAddress().equals(address)) {
+      return;
+    }
+    GrpcChannelEndpoint server = servers.remove(address);
+    if (server != null) {
+      shutdownChannel(server, false);
+    }
+  }
+
+  @Override
+  public void shutdown() {
+    if (!isShutdown.compareAndSet(false, true)) {
+      return;
+    }
+    for (GrpcChannelEndpoint server : servers.values()) {
+      shutdownChannel(server, true);
+    }
+    servers.clear();
+  }
+
+  /**
+   * Shuts down a server's channel.
+   *
+   * <p>First attempts a graceful shutdown. When awaitTermination is true, waits for in-flight RPCs
+   * to complete and forces shutdown on timeout.
+   */
+  private void shutdownChannel(GrpcChannelEndpoint server, boolean awaitTermination) {
+    ManagedChannel channel = server.getChannel();
+    if (channel.isShutdown()) {
+      return;
+    }
+
+    channel.shutdown();
+    if (!awaitTermination) {
+      return;
+    }
+    try {
+      if (!channel.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+        channel.shutdownNow();
+      }
+    } catch (InterruptedException e) {
+      channel.shutdownNow();
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  /** gRPC implementation of {@link ChannelEndpoint}. */
+  static class GrpcChannelEndpoint implements ChannelEndpoint {
+    private final String address;
+    private final ManagedChannel channel;
+
+    /**
+     * Creates a server from a channel provider.
+     *
+     * @param address the server address
+     * @param provider the channel provider (must be a gRPC provider)
+     * @throws IOException if the channel cannot be created
+     */
+    GrpcChannelEndpoint(String address, TransportChannelProvider provider) throws IOException {
+      this.address = address;
+      TransportChannelProvider readyProvider = provider;
+      if (provider.needsHeaders()) {
+        readyProvider = provider.withHeaders(java.util.Collections.emptyMap());
+      }
+      GrpcTransportChannel transportChannel =
+          (GrpcTransportChannel) readyProvider.getTransportChannel();
+      this.channel = (ManagedChannel) transportChannel.getChannel();
+    }
+
+    /**
+     * Creates a server with an existing channel. Primarily for testing.
+     *
+     * @param address the server address
+     * @param channel the managed channel
+     */
+    @VisibleForTesting
+    GrpcChannelEndpoint(String address, ManagedChannel channel) {
+      this.address = address;
+      this.channel = channel;
+    }
+
+    @Override
+    public String getAddress() {
+      return address;
+    }
+
+    @Override
+    public boolean isHealthy() {
+      if (channel.isShutdown() || channel.isTerminated()) {
+        return false;
+      }
+      // Check connectivity state without triggering a connection attempt.
+      // Some channel implementations don't support getState(), in which case
+      // we assume the channel is healthy if it's not shutdown/terminated.
+      try {
+        ConnectivityState state = channel.getState(false);
+        return state != ConnectivityState.SHUTDOWN && state != ConnectivityState.TRANSIENT_FAILURE;
+      } catch (UnsupportedOperationException ignore) {
+        return true;
+      }
+    }
+
+    @Override
+    public ManagedChannel getChannel() {
+      return channel;
+    }
+  }
+}