From 0e31d2cf3cd3df417dc8f9863c8ed60d5b0cea4b Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Tue, 27 Oct 2026 07:55:43 +0000 Subject: [PATCH 1/4] test(spanner): unflake InlineBeginTransactionTest In GrpcResultSet.close(), a previous commit inadvertently moved iterator.close() ahead of listener.onDone(). Cancelling the stream unblocked concurrent reader threads in next() with CANCELLED, triggering yieldError() which completed transactionIdFuture with ABORTED before listener.onDone() could complete it with FAILED_PRECONDITION. This race caused flaky test failures in InlineBeginTransactionTest. This PR fixes the issue by reordering listener.onDone(withBeginTransaction) before iterator.close(), guaranteeing the future is marked FAILED_PRECONDITION before the stream is cancelled. It also includes a deterministic regression test to ensure the fix is robust. --- .../google/cloud/spanner/GrpcResultSet.java | 6 +- .../cloud/spanner/GrpcResultSetTest.java | 91 +++++++++++++++++++ 2 files changed, 96 insertions(+), 1 deletion(-) diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java index 2f3daef47419..ef2cc00a406d 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java @@ -146,8 +146,12 @@ public void close() { closed = true; } boolean withBeginTransaction = iterator.isWithBeginTransaction(); - iterator.close("ResultSet closed"); + // Notify the listener before cancelling the stream. Cancelling the stream causes any thread + // that is concurrently blocked in next() to fail with CANCELLED, which would otherwise race + // with this notification and could fail a pending inline-begin transaction with a different + // error than the one that is caused by closing this ResultSet. listener.onDone(withBeginTransaction); + iterator.close("ResultSet closed"); } @Override diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/GrpcResultSetTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/GrpcResultSetTest.java index 614ad37fa041..6cda8e419842 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/GrpcResultSetTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/GrpcResultSetTest.java @@ -57,7 +57,14 @@ import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import javax.annotation.Nullable; import org.junit.Before; import org.junit.Test; @@ -1306,6 +1313,90 @@ public PartialResultSet next() { assertTrue(listenerReceivedWithBeginTransaction.get()); } + @Test + public void testCloseNotifiesListenerBeforeCancelErrorFromConcurrentReader() throws Exception { + // The first transaction-id outcome that the listener receives wins, just like the + // transactionIdFuture of an inline-begin read/write transaction. + AtomicReference firstOutcome = new AtomicReference<>(); + CountDownLatch readerReportedError = new CountDownLatch(1); + AbstractResultSet.Listener listener = + new NoOpListener() { + @Override + public SpannerException onError( + SpannerException e, boolean withBeginTransaction, boolean lastStatement) { + if (withBeginTransaction) { + firstOutcome.compareAndSet(null, "onError:" + e.getErrorCode()); + } + readerReportedError.countDown(); + return e; + } + + @Override + public void onDone(boolean withBeginTransaction) { + if (withBeginTransaction) { + firstOutcome.compareAndSet(null, "onDone"); + } + } + }; + GrpcStreamIterator iterator = + new GrpcStreamIterator( + /* lastStatement= */ false, 10, /* cancelQueryWhenClientIsClosed= */ false); + SpannerRpc.ResultStreamConsumer streamConsumer = iterator.consumer(); + iterator.setCall( + new SpannerRpc.StreamingCall() { + @Override + public ApiCallContext getCallContext() { + return GrpcCallContext.createDefault(); + } + + @Override + public void cancel(@Nullable String message) { + // Emulate gRPC delivering the CANCELLED status to the stream consumer, and the + // concurrent reader thread processing that error before close() returns. + streamConsumer.onError( + SpannerExceptionFactory.newSpannerException(ErrorCode.CANCELLED, message)); + try { + readerReportedError.await(10, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + @Override + public void request(int numMessages) {} + }, + /* withBeginTransaction= */ true); + GrpcResultSet resultSet = new GrpcResultSet(iterator, listener); + + AtomicReference readerThread = new AtomicReference<>(); + ExecutorService executor = + Executors.newSingleThreadExecutor( + runnable -> { + Thread thread = new Thread(runnable); + readerThread.set(thread); + return thread; + }); + try { + Future nextFuture = executor.submit(resultSet::next); + // Wait until the reader is blocked waiting for the first chunk of the stream. + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (readerThread.get() == null || readerThread.get().getState() != Thread.State.WAITING) { + assertTrue("reader did not block on the stream", System.nanoTime() < deadline); + Thread.yield(); + } + resultSet.close(); + + ExecutionException executionException = + assertThrows(ExecutionException.class, () -> nextFuture.get(10, TimeUnit.SECONDS)); + assertTrue(executionException.getCause() instanceof SpannerException); + assertEquals( + ErrorCode.CANCELLED, ((SpannerException) executionException.getCause()).getErrorCode()); + assertEquals("onDone", firstOutcome.get()); + } finally { + executor.shutdownNow(); + } + } + @Test public void testIsDataAvailable() { assertFalse(resultSet.isDataAvailable()); From 717b619d1276d137cb74ef2ac07bb9e9f6ecb9d8 Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Tue, 27 Oct 2026 08:07:35 +0000 Subject: [PATCH 2/4] Addressed comments --- .../main/java/com/google/cloud/spanner/GrpcResultSet.java | 8 ++++++-- .../java/com/google/cloud/spanner/GrpcResultSetTest.java | 2 +- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java index ef2cc00a406d..496bcaf41071 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java @@ -150,8 +150,12 @@ public void close() { // that is concurrently blocked in next() to fail with CANCELLED, which would otherwise race // with this notification and could fail a pending inline-begin transaction with a different // error than the one that is caused by closing this ResultSet. - listener.onDone(withBeginTransaction); - iterator.close("ResultSet closed"); + try { + listener.onDone(withBeginTransaction); + } finally { + + } + } @Override diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/GrpcResultSetTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/GrpcResultSetTest.java index 6cda8e419842..f092385b2db8 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/GrpcResultSetTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/GrpcResultSetTest.java @@ -1382,7 +1382,7 @@ public void request(int numMessages) {} long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); while (readerThread.get() == null || readerThread.get().getState() != Thread.State.WAITING) { assertTrue("reader did not block on the stream", System.nanoTime() < deadline); - Thread.yield(); + Thread.sleep(10); } resultSet.close(); From 3d692495f27c0ab08ef5d8d0eeef55457cc7a438 Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Tue, 27 Oct 2026 08:08:45 +0000 Subject: [PATCH 3/4] Addressed comments --- .../src/main/java/com/google/cloud/spanner/GrpcResultSet.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java index 496bcaf41071..5aa14c550ae6 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/GrpcResultSet.java @@ -153,9 +153,8 @@ public void close() { try { listener.onDone(withBeginTransaction); } finally { - + iterator.close("ResultSet closed"); } - } @Override From 287394bf2697cf27505805e84f2323ac2e633b92 Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Tue, 27 Oct 2026 08:19:56 +0000 Subject: [PATCH 4/4] Addressed comments --- .../test/java/com/google/cloud/spanner/GrpcResultSetTest.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/GrpcResultSetTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/GrpcResultSetTest.java index f092385b2db8..ec087f58bfd2 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/GrpcResultSetTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/GrpcResultSetTest.java @@ -1381,6 +1381,10 @@ public void request(int numMessages) {} // Wait until the reader is blocked waiting for the first chunk of the stream. long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); while (readerThread.get() == null || readerThread.get().getState() != Thread.State.WAITING) { + if (nextFuture.isDone()) { + nextFuture.get(); + throw new AssertionError("Reader thread terminated unexpectedly without blocking"); + } assertTrue("reader did not block on the stream", System.nanoTime() < deadline); Thread.sleep(10); }