Skip to content
This repository was archived by the owner on Apr 7, 2026. It is now read-only.

Commit 3d8a02c

Browse files
rahul2393rahul2393
committed
incorporate changes
1 parent 97bd6cf commit 3d8a02c

5 files changed

Lines changed: 166 additions & 166 deletions

File tree

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

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,15 +23,15 @@
2323
* Represents a Spanner server endpoint for location-aware routing.
2424
*
2525
* <p>Each instance wraps a gRPC {@link ManagedChannel} connected to a specific Spanner server. The
26-
* {@link ChannelFinderServerFactory} creates and caches these instances.
26+
* {@link ChannelEndpointCache} creates and caches these instances.
2727
*
2828
* <p>Implementations must be thread-safe as instances may be shared across multiple concurrent
2929
* operations.
3030
*
31-
* @see ChannelFinderServerFactory
31+
* @see ChannelEndpointCache
3232
*/
3333
@InternalApi
34-
public interface ChannelFinderServer {
34+
public interface ChannelEndpoint {
3535

3636
/**
3737
* Returns the network address of this server.
@@ -57,8 +57,8 @@ public interface ChannelFinderServer {
5757
/**
5858
* Returns the gRPC channel for making RPCs to this server.
5959
*
60-
* <p>The returned channel is managed by the {@link ChannelFinderServerFactory} and should not be
61-
* shut down directly by callers.
60+
* <p>The returned channel is managed by the {@link ChannelEndpointCache} and should not be shut
61+
* down directly by callers.
6262
*
6363
* @return the managed channel for this server
6464
*/

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

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -19,31 +19,31 @@
1919
import com.google.api.core.InternalApi;
2020

2121
/**
22-
* Factory for creating and caching server connections for location-aware routing.
22+
* Cache for server connections used in location-aware routing.
2323
*
24-
* <p>Implementations are expected to cache {@link ChannelFinderServer} instances such that repeated
24+
* <p>Implementations are expected to cache {@link ChannelEndpoint} instances such that repeated
2525
* calls with the same address return the same instance. This allows routing components to
2626
* efficiently manage server references.
2727
*
2828
* <p>Implementations must be thread-safe. Multiple threads may concurrently call {@link
29-
* #create(String)} with different addresses.
29+
* #get(String)} with different addresses.
3030
*/
3131
@InternalApi
32-
public interface ChannelFinderServerFactory {
32+
public interface ChannelEndpointCache {
3333

3434
/**
35-
* Returns the default server endpoint.
35+
* Returns the default channel endpoint.
3636
*
37-
* <p>The default server is the original endpoint configured in {@link
37+
* <p>The default channel is the original endpoint configured in {@link
3838
* com.google.cloud.spanner.SpannerOptions}. It is used as a fallback when the location cache does
3939
* not have routing information for a request.
4040
*
41-
* @return the default server, never null
41+
* @return the default channel, never null
4242
*/
43-
ChannelFinderServer defaultServer();
43+
ChannelEndpoint defaultChannel();
4444

4545
/**
46-
* Creates or retrieves a cached server for the given address.
46+
* Returns a cached server for the given address, creating it if needed.
4747
*
4848
* <p>If a server for this address already exists in the cache, the cached instance is returned.
4949
* Otherwise, a new server connection is created and cached.
@@ -52,7 +52,7 @@ public interface ChannelFinderServerFactory {
5252
* @return a server instance for the address, never null
5353
* @throws com.google.cloud.spanner.SpannerException if the channel cannot be created
5454
*/
55-
ChannelFinderServer create(String address);
55+
ChannelEndpoint get(String address);
5656

5757
/**
5858
* Evicts a server from the cache and gracefully shuts down its channel.
@@ -73,7 +73,7 @@ public interface ChannelFinderServerFactory {
7373
* <p>This method should be called when the Spanner client is closed to release all resources.
7474
* Each channel is shut down gracefully, allowing in-flight RPCs to complete.
7575
*
76-
* <p>After calling this method, the factory should not be used to create new connections.
76+
* <p>After calling this method, the cache should not be used to create new connections.
7777
*/
7878
void shutdown();
7979
}

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

Lines changed: 37 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -29,50 +29,51 @@
2929
import java.util.Map;
3030
import java.util.concurrent.ConcurrentHashMap;
3131
import java.util.concurrent.TimeUnit;
32+
import java.util.concurrent.atomic.AtomicBoolean;
3233

3334
/**
34-
* gRPC implementation of {@link ChannelFinderServerFactory}.
35+
* gRPC implementation of {@link ChannelEndpointCache}.
3536
*
36-
* <p>This factory creates and caches gRPC channels per address. It uses {@link
37+
* <p>This cache creates and caches gRPC channels per address. It uses {@link
3738
* InstantiatingGrpcChannelProvider#withEndpoint(String)} to create new channels with the same
3839
* configuration but different endpoints, avoiding race conditions.
3940
*/
4041
@InternalApi
41-
class GrpcChannelFinderServerFactory implements ChannelFinderServerFactory {
42+
class GrpcChannelEndpointCache implements ChannelEndpointCache {
4243

4344
/** Timeout for graceful channel shutdown. */
4445
private static final long SHUTDOWN_TIMEOUT_SECONDS = 5;
4546

4647
private final InstantiatingGrpcChannelProvider baseProvider;
47-
private final Map<String, GrpcChannelFinderServer> servers = new ConcurrentHashMap<>();
48-
private final GrpcChannelFinderServer defaultServer;
49-
private volatile boolean isShutdown = false;
48+
private final Map<String, GrpcChannelEndpoint> servers = new ConcurrentHashMap<>();
49+
private final GrpcChannelEndpoint defaultEndpoint;
50+
private final AtomicBoolean isShutdown = new AtomicBoolean(false);
5051

5152
/**
52-
* Creates a new factory with the given channel provider.
53+
* Creates a new cache with the given channel provider.
5354
*
5455
* @param channelProvider the base provider used to create channels. New channels for different
5556
* endpoints are created using {@link InstantiatingGrpcChannelProvider#withEndpoint(String)}.
5657
* @throws IOException if the default channel cannot be created
5758
*/
58-
public GrpcChannelFinderServerFactory(InstantiatingGrpcChannelProvider channelProvider)
59+
public GrpcChannelEndpointCache(InstantiatingGrpcChannelProvider channelProvider)
5960
throws IOException {
6061
this.baseProvider = channelProvider;
6162
String defaultEndpoint = channelProvider.getEndpoint();
62-
this.defaultServer = new GrpcChannelFinderServer(defaultEndpoint, channelProvider);
63-
this.servers.put(defaultEndpoint, this.defaultServer);
63+
this.defaultEndpoint = new GrpcChannelEndpoint(defaultEndpoint, channelProvider);
64+
this.servers.put(defaultEndpoint, this.defaultEndpoint);
6465
}
6566

6667
@Override
67-
public ChannelFinderServer defaultServer() {
68-
return defaultServer;
68+
public ChannelEndpoint defaultChannel() {
69+
return defaultEndpoint;
6970
}
7071

7172
@Override
72-
public ChannelFinderServer create(String address) {
73-
if (isShutdown) {
73+
public ChannelEndpoint get(String address) {
74+
if (isShutdown.get()) {
7475
throw SpannerExceptionFactory.newSpannerException(
75-
ErrorCode.FAILED_PRECONDITION, "ChannelFinderServerFactory has been shut down");
76+
ErrorCode.FAILED_PRECONDITION, "ChannelEndpointCache has been shut down");
7677
}
7778

7879
return servers.computeIfAbsent(
@@ -82,7 +83,7 @@ public ChannelFinderServer create(String address) {
8283
// Create a new provider with the same config but different endpoint.
8384
// This is thread-safe as withEndpoint() returns a new provider instance.
8485
TransportChannelProvider newProvider = baseProvider.withEndpoint(addr);
85-
return new GrpcChannelFinderServer(addr, newProvider);
86+
return new GrpcChannelEndpoint(addr, newProvider);
8687
} catch (IOException e) {
8788
throw SpannerExceptionFactory.newSpannerException(
8889
ErrorCode.INTERNAL, "Failed to create channel for address: " + addr, e);
@@ -92,37 +93,42 @@ public ChannelFinderServer create(String address) {
9293

9394
@Override
9495
public void evict(String address) {
95-
if (defaultServer.getAddress().equals(address)) {
96+
if (defaultEndpoint.getAddress().equals(address)) {
9697
return;
9798
}
98-
GrpcChannelFinderServer server = servers.remove(address);
99+
GrpcChannelEndpoint server = servers.remove(address);
99100
if (server != null) {
100-
shutdownServerGracefully(server);
101+
shutdownChannel(server, false);
101102
}
102103
}
103104

104105
@Override
105106
public void shutdown() {
106-
isShutdown = true;
107-
for (GrpcChannelFinderServer server : servers.values()) {
108-
shutdownServerGracefully(server);
107+
if (!isShutdown.compareAndSet(false, true)) {
108+
return;
109+
}
110+
for (GrpcChannelEndpoint server : servers.values()) {
111+
shutdownChannel(server, true);
109112
}
110113
servers.clear();
111114
}
112115

113116
/**
114-
* Gracefully shuts down a server's channel.
117+
* Shuts down a server's channel.
115118
*
116-
* <p>First attempts a graceful shutdown, waiting for in-flight RPCs to complete. If the timeout
117-
* is exceeded, forces immediate shutdown.
119+
* <p>First attempts a graceful shutdown. When awaitTermination is true, waits for in-flight RPCs
120+
* to complete and forces shutdown on timeout.
118121
*/
119-
private void shutdownServerGracefully(GrpcChannelFinderServer server) {
122+
private void shutdownChannel(GrpcChannelEndpoint server, boolean awaitTermination) {
120123
ManagedChannel channel = server.getChannel();
121124
if (channel.isShutdown()) {
122125
return;
123126
}
124127

125128
channel.shutdown();
129+
if (!awaitTermination) {
130+
return;
131+
}
126132
try {
127133
if (!channel.awaitTermination(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
128134
channel.shutdownNow();
@@ -133,8 +139,8 @@ private void shutdownServerGracefully(GrpcChannelFinderServer server) {
133139
}
134140
}
135141

136-
/** gRPC implementation of {@link ChannelFinderServer}. */
137-
static class GrpcChannelFinderServer implements ChannelFinderServer {
142+
/** gRPC implementation of {@link ChannelEndpoint}. */
143+
static class GrpcChannelEndpoint implements ChannelEndpoint {
138144
private final String address;
139145
private final ManagedChannel channel;
140146

@@ -145,7 +151,7 @@ static class GrpcChannelFinderServer implements ChannelFinderServer {
145151
* @param provider the channel provider (must be a gRPC provider)
146152
* @throws IOException if the channel cannot be created
147153
*/
148-
GrpcChannelFinderServer(String address, TransportChannelProvider provider) throws IOException {
154+
GrpcChannelEndpoint(String address, TransportChannelProvider provider) throws IOException {
149155
this.address = address;
150156
TransportChannelProvider readyProvider = provider;
151157
if (provider.needsHeaders()) {
@@ -163,7 +169,7 @@ static class GrpcChannelFinderServer implements ChannelFinderServer {
163169
* @param channel the managed channel
164170
*/
165171
@VisibleForTesting
166-
GrpcChannelFinderServer(String address, ManagedChannel channel) {
172+
GrpcChannelEndpoint(String address, ManagedChannel channel) {
167173
this.address = address;
168174
this.channel = channel;
169175
}
@@ -184,7 +190,7 @@ public boolean isHealthy() {
184190
try {
185191
ConnectivityState state = channel.getState(false);
186192
return state != ConnectivityState.SHUTDOWN && state != ConnectivityState.TRANSIENT_FAILURE;
187-
} catch (UnsupportedOperationException e) {
193+
} catch (UnsupportedOperationException ignore) {
188194
return true;
189195
}
190196
}

‎google-cloud-spanner/src/test/java/com/google/cloud/spanner/spi/v1/GrpcChannelEndpointCacheTest.java‎

Lines changed: 113 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,113 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.google.cloud.spanner.spi.v1;
18+
19+
import static com.google.common.truth.Truth.assertThat;
20+
import static org.junit.Assert.assertThrows;
21+
22+
import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
23+
import com.google.cloud.spanner.SpannerException;
24+
import io.grpc.ManagedChannelBuilder;
25+
import org.junit.Test;
26+
27+
public class GrpcChannelEndpointCacheTest {
28+
29+
private static InstantiatingGrpcChannelProvider createProvider(String endpoint) {
30+
return InstantiatingGrpcChannelProvider.newBuilder()
31+
.setEndpoint(endpoint)
32+
.setChannelConfigurator(ManagedChannelBuilder::usePlaintext)
33+
.build();
34+
}
35+
36+
@Test
37+
public void defaultChannelIsCached() throws Exception {
38+
GrpcChannelEndpointCache cache = new GrpcChannelEndpointCache(createProvider("localhost:1234"));
39+
try {
40+
ChannelEndpoint defaultChannel = cache.defaultChannel();
41+
ChannelEndpoint server = cache.get(defaultChannel.getAddress());
42+
assertThat(server).isSameInstanceAs(defaultChannel);
43+
} finally {
44+
cache.shutdown();
45+
}
46+
}
47+
48+
@Test
49+
public void getCachesPerAddress() throws Exception {
50+
GrpcChannelEndpointCache cache = new GrpcChannelEndpointCache(createProvider("localhost:1234"));
51+
try {
52+
ChannelEndpoint first = cache.get("localhost:1111");
53+
ChannelEndpoint second = cache.get("localhost:1111");
54+
ChannelEndpoint third = cache.get("localhost:2222");
55+
56+
assertThat(second).isSameInstanceAs(first);
57+
assertThat(third).isNotSameInstanceAs(first);
58+
} finally {
59+
cache.shutdown();
60+
}
61+
}
62+
63+
@Test
64+
public void evictRemovesNonDefaultServer() throws Exception {
65+
GrpcChannelEndpointCache cache = new GrpcChannelEndpointCache(createProvider("localhost:1234"));
66+
try {
67+
ChannelEndpoint first = cache.get("localhost:1111");
68+
cache.evict("localhost:1111");
69+
ChannelEndpoint second = cache.get("localhost:1111");
70+
71+
assertThat(second).isNotSameInstanceAs(first);
72+
} finally {
73+
cache.shutdown();
74+
}
75+
}
76+
77+
@Test
78+
public void evictIgnoresDefaultChannel() throws Exception {
79+
GrpcChannelEndpointCache cache = new GrpcChannelEndpointCache(createProvider("localhost:1234"));
80+
try {
81+
ChannelEndpoint defaultChannel = cache.defaultChannel();
82+
cache.evict(defaultChannel.getAddress());
83+
ChannelEndpoint server = cache.get(defaultChannel.getAddress());
84+
85+
assertThat(server).isSameInstanceAs(defaultChannel);
86+
} finally {
87+
cache.shutdown();
88+
}
89+
}
90+
91+
@Test
92+
public void shutdownPreventsNewServers() throws Exception {
93+
GrpcChannelEndpointCache cache = new GrpcChannelEndpointCache(createProvider("localhost:1234"));
94+
cache.shutdown();
95+
96+
assertThrows(SpannerException.class, () -> cache.get("localhost:1111"));
97+
assertThat(cache.defaultChannel().getChannel().isShutdown()).isTrue();
98+
}
99+
100+
@Test
101+
public void healthReflectsChannelShutdown() throws Exception {
102+
GrpcChannelEndpointCache cache = new GrpcChannelEndpointCache(createProvider("localhost:1234"));
103+
try {
104+
ChannelEndpoint server = cache.get("localhost:1111");
105+
assertThat(server.isHealthy()).isTrue();
106+
107+
server.getChannel().shutdownNow();
108+
assertThat(server.isHealthy()).isFalse();
109+
} finally {
110+
cache.shutdown();
111+
}
112+
}
113+
}

0 commit comments

Comments
 (0)