From 1d1fe47a63c03e6e067c75068918b4aed7e6f2f2 Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Mon, 2 Nov 2026 14:19:04 +0000 Subject: [PATCH 1/2] test(spanner): unflake ReadWriteTransactionWithInlineBeginTest Synchronize on the prefetch request() call instead of ExecuteSqlRequest mock server receipt to prevent close() from racing with stream registration. --- .../ReadWriteTransactionWithInlineBeginTest.java | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ReadWriteTransactionWithInlineBeginTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ReadWriteTransactionWithInlineBeginTest.java index b5591d3072a8..b1deeeda2965 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ReadWriteTransactionWithInlineBeginTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ReadWriteTransactionWithInlineBeginTest.java @@ -1179,6 +1179,13 @@ public void closeBeforeTransactionIdReturnedDoesNotRaceWithCancelledConsumer() t // closed, so onError does not overwrite transactionIdFuture with a simulated ABORTED before // onDone(true) sets FAILED_PRECONDITION. CountDownLatch consumerFinishedNextLatch = new CountDownLatch(1); + // Counted down when the initial prefetch request is issued on the call. This only happens after + // startStream() has returned and the stream has been registered with the result set iterator, + // which guarantees that a subsequent close() cancels the call from the closing thread. Waiting + // only for the ExecuteSqlRequest to reach the server is not sufficient, as close() could then + // run before the stream is registered, in which case the call is cancelled later on the + // consumer thread itself, which would block in cancel() below waiting for itself. + CountDownLatch streamRegisteredLatch = new CountDownLatch(1); SpannerOptions options = createSpannerOptionsWithStreamInterceptors( (invocation, delegate) -> { @@ -1202,6 +1209,7 @@ public void cancel(String message) { @Override public void request(int numMessages) { delegateCall.request(numMessages); + streamRegisteredLatch.countDown(); } }; }, @@ -1230,11 +1238,7 @@ public void request(int numMessages) { consumerFinishedNextLatch.countDown(); } }); - try { - mockSpanner.waitForRequestsToContain(ExecuteSqlRequest.class, 5000); - } catch (Exception exceptionFromWait) { - throw new RuntimeException(exceptionFromWait); - } + assertTrue(streamRegisteredLatch.await(5, TimeUnit.SECONDS)); resultSet.close(); mockSpanner.unfreeze(); ExecutionException consumerExecutionException = From b7970142b6be3ca8fa9bbb74a91fd044f9364519 Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Mon, 2 Nov 2026 15:52:48 +0000 Subject: [PATCH 2/2] Addressed comments --- ...adWriteTransactionWithInlineBeginTest.java | 20 +++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ReadWriteTransactionWithInlineBeginTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ReadWriteTransactionWithInlineBeginTest.java index b1deeeda2965..59e23b54df4e 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ReadWriteTransactionWithInlineBeginTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/ReadWriteTransactionWithInlineBeginTest.java @@ -1181,11 +1181,10 @@ public void closeBeforeTransactionIdReturnedDoesNotRaceWithCancelledConsumer() t CountDownLatch consumerFinishedNextLatch = new CountDownLatch(1); // Counted down when the initial prefetch request is issued on the call. This only happens after // startStream() has returned and the stream has been registered with the result set iterator, - // which guarantees that a subsequent close() cancels the call from the closing thread. Waiting - // only for the ExecuteSqlRequest to reach the server is not sufficient, as close() could then - // run before the stream is registered, in which case the call is cancelled later on the - // consumer thread itself, which would block in cancel() below waiting for itself. + // which guarantees that a subsequent close() cancels the call from the closing thread rather + // than running before stream registration. CountDownLatch streamRegisteredLatch = new CountDownLatch(1); + AtomicReference consumerThread = new AtomicReference<>(); SpannerOptions options = createSpannerOptionsWithStreamInterceptors( (invocation, delegate) -> { @@ -1199,10 +1198,14 @@ public ApiCallContext getCallContext() { @Override public void cancel(String message) { delegateCall.cancel(message); - try { - assertTrue(consumerFinishedNextLatch.await(5, TimeUnit.SECONDS)); - } catch (InterruptedException interruptedException) { - Thread.currentThread().interrupt(); + // Only wait if called from an external thread (e.g. resultSet.close()), + // not if cancel() is invoked by the consumer thread itself. + if (Thread.currentThread() != consumerThread.get()) { + try { + assertTrue(consumerFinishedNextLatch.await(5, TimeUnit.SECONDS)); + } catch (InterruptedException interruptedException) { + Thread.currentThread().interrupt(); + } } } @@ -1232,6 +1235,7 @@ public void request(int numMessages) { Future nextFuture = executor.submit( () -> { + consumerThread.set(Thread.currentThread()); try { resultSet.next(); } finally {