Skip to content

test(spanner): unflake LocationAwareSharedBackendReplicaHarnessTest #14396

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sakthivelmanii merged 1 commit into from
Sep 16, 2026
+46 −0
Merged

test(spanner): unflake LocationAwareSharedBackendReplicaHarnessTest #14396

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .../src/test/java/com/google/cloud/spanner/LocationAwareSharedBackendReplicaHarnessTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

package com.google.cloud.spanner;

import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
Expand Down Expand Up @@ -49,6 +50,7 @@
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.protobuf.ProtoUtils;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
Expand Down Expand Up @@ -573,8 +575,13 @@ private static boolean drain(ResultSet resultSet) {
return sawRow;
}

private static void waitForAllReplicasConnected(SharedBackendReplicaHarness harness) {
await().atMost(Duration.ofSeconds(10)).until(harness::allReplicasConnected);
}

private static int waitForReplicaRoutedRead(
DatabaseClient client, SharedBackendReplicaHarness harness) throws InterruptedException {
waitForAllReplicasConnected(harness);
long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
while (System.nanoTime() < deadlineNanos) {
try (ResultSet resultSet =
Expand Down
39 changes: 39 additions & 0 deletions ...gle-cloud-spanner/src/test/java/com/google/cloud/spanner/SharedBackendReplicaHarness.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,12 +34,14 @@
import com.google.spanner.v1.Session;
import com.google.spanner.v1.SpannerGrpc;
import com.google.spanner.v1.Transaction;
import io.grpc.Attributes;
import io.grpc.Metadata;
import io.grpc.Server;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import io.grpc.ServerInterceptor;
import io.grpc.ServerInterceptors;
import io.grpc.ServerTransportFilter;
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
import io.grpc.stub.StreamObserver;
import java.io.Closeable;
Expand All @@ -50,6 +52,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

/** Shared-backend replica harness for end-to-end location-aware routing tests. */
final class SharedBackendReplicaHarness implements Closeable {
Expand All @@ -71,11 +74,24 @@ static final class HookedReplicaSpannerService extends SpannerGrpc.SpannerImplBa
private final Map<String, ArrayDeque<Throwable>> methodErrors = new HashMap<>();
private final Map<String, List<AbstractMessage>> requests = new HashMap<>();
private final Map<String, List<String>> requestIds = new HashMap<>();
private final AtomicInteger activeConnections = new AtomicInteger();

private HookedReplicaSpannerService(MockSpannerServiceImpl backend) {
this.backend = backend;
}

void recordConnectionReady() {
activeConnections.incrementAndGet();
}

void recordConnectionTerminated() {
activeConnections.decrementAndGet();
}
Comment on lines +87 to +89

@gemini-code-assist gemini-code-assist Bot Sep 16, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Under certain network conditions or connection handshake failures, gRPC's transportTerminated can be invoked without a corresponding successful transportReady call. If this occurs, activeConnections could decrement below zero, causing subsequent successful connections to not be recognized correctly (since activeConnections.get() > 0 would remain false or delayed). To make the connection tracking robust against such anomalies, consider using updateAndGet to prevent the counter from dropping below zero.

Suggested change
void recordConnectionTerminated() {
activeConnections.decrementAndGet();
}
void recordConnectionTerminated() {
activeConnections.updateAndGet(val -> Math.max(0, val - 1));
}


boolean hasConnected() {
return activeConnections.get() > 0;
}

synchronized void putMethodErrors(String method, Throwable... errors) {
ArrayDeque<Throwable> queue = new ArrayDeque<>();
for (Throwable error : errors) {
Expand Down Expand Up @@ -278,12 +294,35 @@ public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
Server server =
NettyServerBuilder.forAddress(address)
.addService(ServerInterceptors.intercept(service, interceptor))
.addTransportFilter(
new ServerTransportFilter() {
@Override
public Attributes transportReady(Attributes transportAttrs) {
service.recordConnectionReady();
return super.transportReady(transportAttrs);
}

@Override
public void transportTerminated(Attributes transportAttrs) {
service.recordConnectionTerminated();
super.transportTerminated(transportAttrs);
}
})
.build()
.start();
servers.add(server);
return "localhost:" + server.getPort();
}

boolean allReplicasConnected() {
for (HookedReplicaSpannerService replica : replicas) {
if (!replica.hasConnected()) {
return false;
}
}
return true;
}

void clearRequests() {
defaultReplica.clearRequests();
for (HookedReplicaSpannerService replica : replicas) {
Expand Down
Loading