From c1900e5ab77333124ddbc7edcbf0cdc38ef3af04 Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Tue, 27 Oct 2026 11:51:19 +0000 Subject: [PATCH 1/6] test(spanner): unflake ConnectionAsyncApiAbortedTest Fixes a client race condition in the Connection API where an unawaited asynchronous statement aborts and fails with AbortedDueToConcurrentModificationException before the application thread calls commitAsync(). ConnectionImpl.hasTransactionalChanges() incorrectly returned false because it required the state to be STARTED, causing commitAsync() to take the 'nothing to commit' no-op branch and silently swallow the aborted exception. Additionally, commitAsync()'s thread could overwrite the ABORTED state with COMMITTING, bypassing the aborted exception check. This fix scopes the aborted transaction check to commit only, avoiding alterations to public API behavior, updates checkAborted() to key on the non-null abortedException, and adds a regression test. --- .../spanner/connection/ConnectionImpl.java | 14 ++++++++- .../connection/ReadWriteTransaction.java | 4 ++- .../ConnectionAsyncApiAbortedTest.java | 30 +++++++++++++++++++ 3 files changed, 46 insertions(+), 2 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionImpl.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionImpl.java index cadd63757398..3e191023c3b5 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionImpl.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionImpl.java @@ -1142,6 +1142,17 @@ private boolean hasTransactionalChanges() { return internalIsTransactionStarted() || this.connectionState.hasTransactionalChanges(); } + /** + * Returns true if the current transaction has been aborted and could not be retried. Committing + * such a transaction must return the abort error instead of silently succeeding. This can happen + * when a statement that was executed asynchronously aborts before commit is called. + */ + private boolean isCurrentTransactionAborted() { + return internalIsInTransaction() + && this.currentUnitOfWork != null + && this.currentUnitOfWork.getState() == UnitOfWorkState.ABORTED; + } + @Override public Timestamp getReadTimestamp() { ConnectionPreconditions.checkState(!isClosed(), CLOSED_ERROR_MSG); @@ -1387,7 +1398,8 @@ private ApiFuture endCurrentTransactionAsync( statementTag == null, "Statement tags are not supported for COMMIT or ROLLBACK"); ApiFuture res; try { - if (hasTransactionalChanges()) { + if (hasTransactionalChanges() + || (endTransactionMethod == commit && isCurrentTransactionAborted())) { res = endTransactionMethod.endAsync( callType, getCurrentUnitOfWorkOrStartNewUnitOfWork(parsedStatement)); diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java index 246c203c8674..6caee0fb3f32 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java @@ -601,7 +601,9 @@ public boolean isActive() { } void checkAborted() { - if (this.state == UnitOfWorkState.ABORTED && this.abortedException != null) { + // Only check abortedException and not the state, as the ABORTED state can be overwritten by a + // commit that is called while (or after) an asynchronous statement aborted the transaction. + if (this.abortedException != null) { if (this.abortedException instanceof AbortedDueToConcurrentModificationException) { throw SpannerExceptionFactory.newAbortedDueToConcurrentModificationException( (AbortedDueToConcurrentModificationException) this.abortedException); diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java index 1f4378ecd86b..a3e26b71c962 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java @@ -531,6 +531,36 @@ public void testBlindUpdateAborted_WithConcurrentModification() { } } + @Test + public void testCommitAfterBlindUpdateAborted_WithConcurrentModification() { + Statement update1 = Statement.of("UPDATE FOO SET BAR=1 WHERE BAZ=100"); + mockSpanner.putStatementResult(StatementResult.update(update1, 100)); + + RetryCounter counter = new RetryCounter(); + try (Connection connection = createConnection(counter)) { + get(connection.executeUpdateAsync(update1)); + mockSpanner.putStatementResult(StatementResult.update(update1, 200)); + + mockSpanner.abortNextStatement(); + ApiFuture insert = connection.executeUpdateAsync(INSERT_STATEMENT); + // Wait until the failed retry has marked the transaction as aborted before calling commit. + try { + get(insert); + fail("Missing expected exception"); + } catch (AbortedDueToConcurrentModificationException e) { + assertThat(counter.retryCount).isEqualTo(1); + } + + // Committing a transaction that has been aborted must fail and must not silently succeed. + try { + get(connection.commitAsync()); + fail("Missing expected exception"); + } catch (AbortedDueToConcurrentModificationException e) { + assertThat(counter.retryCount).isEqualTo(1); + } + } + } + @Test public void testMultipleBlindUpdatesAborted_WithConcurrentModification() { Statement update1 = Statement.of("UPDATE FOO SET BAR=1 WHERE BAZ=100"); From 2b7425ef736fd4a3554a4abd27a361905cafb20a Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Tue, 27 Oct 2026 15:29:44 +0000 Subject: [PATCH 2/6] Addressed comments --- .../connection/ReadWriteTransaction.java | 5 +- .../ConnectionAsyncApiAbortedTest.java | 93 ++++++++++--------- 2 files changed, 51 insertions(+), 47 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java index 6caee0fb3f32..bd490e60296b 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java @@ -1329,7 +1329,7 @@ private void invokeTransactionRetryListenersOnFinish(RetryResult result) { @Override public Void call() { try { - if (state != UnitOfWorkState.ABORTED && rolledBackToSavepointException == null) { + if (state != UnitOfWorkState.ABORTED && abortedException == null && rolledBackToSavepointException == null) { // Make sure the transaction has actually started before we try to rollback. get(txContextFuture); txManager.rollback(); @@ -1358,10 +1358,11 @@ private ApiFuture rollbackAsync(CallType callType, boolean updateStatusAnd state == UnitOfWorkState.STARTED || state == UnitOfWorkState.ABORTED, "This transaction has status " + state.name()); cancelScheduledKeepAlivePing(); + boolean isAborted = (state == UnitOfWorkState.ABORTED) || (this.abortedException != null); if (updateStatusAndEndSpan) { state = UnitOfWorkState.ROLLED_BACK; } - if (txContextFuture != null && state != UnitOfWorkState.ABORTED) { + if (txContextFuture != null && !isAborted) { ApiFuture result = executeStatementAsync( callType, ROLLBACK_STATEMENT, rollbackCallable, SpannerGrpc.getRollbackMethod()); diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java index a3e26b71c962..147a3d9f8c60 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java @@ -19,6 +19,9 @@ import static com.google.cloud.spanner.SpannerApiFutures.get; import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.fail; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; + import com.google.api.core.ApiFuture; import com.google.api.core.SettableApiFuture; @@ -138,15 +141,15 @@ protected Builder configureConnectionOptions(Builder builder) { public void testSingleQueryAborted() { RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertThat(counter.retryCount).isEqualTo(0); + assertEquals(0, counter.retryCount); mockSpanner.setExecuteStreamingSqlExecutionTime( SimulatedExecutionTime.ofException( mockSpanner.createAbortedException(ByteString.copyFromUtf8("test")))); QueryResult res = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT); assertThat(get(res.finished)).isNull(); - assertThat(res.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(RANDOM_RESULT_SET_ROW_COUNT, res.rowCount.get()); + assertEquals(1, counter.retryCount); } } @@ -154,7 +157,7 @@ public void testSingleQueryAborted() { public void testTwoQueriesSecondAborted() { RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertThat(counter.retryCount).isEqualTo(0); + assertEquals(0, counter.retryCount); QueryResult res1 = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT); mockSpanner.setExecuteStreamingSqlExecutionTime( SimulatedExecutionTime.ofException( @@ -162,10 +165,10 @@ public void testTwoQueriesSecondAborted() { QueryResult res2 = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT_2); assertThat(get(res1.finished)).isNull(); - assertThat(res1.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); + assertEquals(RANDOM_RESULT_SET_ROW_COUNT, res1.rowCount.get()); assertThat(get(res2.finished)).isNull(); - assertThat(res2.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT_2); - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(RANDOM_RESULT_SET_ROW_COUNT_2, res2.rowCount.get()); + assertEquals(1, counter.retryCount); } } @@ -173,7 +176,7 @@ public void testTwoQueriesSecondAborted() { public void testTwoQueriesBothAborted() throws InterruptedException { RetryCounter counter = new RetryCounter(1); try (Connection connection = createConnection(counter)) { - assertThat(counter.retryCount).isEqualTo(0); + assertEquals(0, counter.retryCount); mockSpanner.setExecuteStreamingSqlExecutionTime( SimulatedExecutionTime.ofException( mockSpanner.createAbortedException(ByteString.copyFromUtf8("test")))); @@ -186,10 +189,10 @@ public void testTwoQueriesBothAborted() throws InterruptedException { QueryResult res2 = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT_2); assertThat(get(res1.finished)).isNull(); - assertThat(res1.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); + assertEquals(RANDOM_RESULT_SET_ROW_COUNT, res1.rowCount.get()); assertThat(get(res2.finished)).isNull(); - assertThat(res2.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT_2); - assertThat(counter.retryCount).isEqualTo(2); + assertEquals(RANDOM_RESULT_SET_ROW_COUNT_2, res2.rowCount.get()); + assertEquals(2, counter.retryCount); } } @@ -201,12 +204,12 @@ public void testSingleQueryAbortedMidway() { RANDOM_RESULT_SET_ROW_COUNT / 2)); RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertThat(counter.retryCount).isEqualTo(0); + assertEquals(0, counter.retryCount); QueryResult res = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT); assertThat(get(res.finished)).isNull(); - assertThat(res.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(RANDOM_RESULT_SET_ROW_COUNT, res.rowCount.get()); + assertEquals(1, counter.retryCount); } } @@ -214,7 +217,7 @@ public void testSingleQueryAbortedMidway() { public void testTwoQueriesSecondAbortedMidway() { RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertThat(counter.retryCount).isEqualTo(0); + assertEquals(0, counter.retryCount); QueryResult res1 = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT); mockSpanner.setExecuteStreamingSqlExecutionTime( SimulatedExecutionTime.ofStreamException( @@ -223,10 +226,10 @@ public void testTwoQueriesSecondAbortedMidway() { QueryResult res2 = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT_2); assertThat(get(res1.finished)).isNull(); - assertThat(res1.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); + assertEquals(RANDOM_RESULT_SET_ROW_COUNT, res1.rowCount.get()); assertThat(get(res2.finished)).isNull(); - assertThat(res2.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT_2); - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(RANDOM_RESULT_SET_ROW_COUNT_2, res2.rowCount.get()); + assertEquals(1, counter.retryCount); } } @@ -238,7 +241,7 @@ public void testTwoQueriesOneAbortedMidway() { Math.min(RANDOM_RESULT_SET_ROW_COUNT / 2, RANDOM_RESULT_SET_ROW_COUNT_2 / 2))); RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertThat(counter.retryCount).isEqualTo(0); + assertEquals(0, counter.retryCount); // These AsyncResultSets will be consumed in parallel. One of them will (at random) abort // halfway. QueryResult res1 = @@ -247,10 +250,10 @@ public void testTwoQueriesOneAbortedMidway() { executeQueryAsync(connection, SELECT_RANDOM_STATEMENT_2, multiThreadedExecutor); assertThat(get(res1.finished)).isNull(); - assertThat(res1.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); + assertEquals(RANDOM_RESULT_SET_ROW_COUNT, res1.rowCount.get()); assertThat(get(res2.finished)).isNull(); - assertThat(res2.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT_2); - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(RANDOM_RESULT_SET_ROW_COUNT_2, res2.rowCount.get()); + assertEquals(1, counter.retryCount); } } @@ -262,7 +265,7 @@ public void testUpdateAndQueryAbortedMidway() throws InterruptedException { RANDOM_RESULT_SET_ROW_COUNT / 2)); final RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertThat(counter.retryCount).isEqualTo(0); + assertEquals(0, counter.retryCount); final SettableApiFuture rowCount = SettableApiFuture.create(); final CountDownLatch updateLatch = new CountDownLatch(1); final CountDownLatch queryLatch = new CountDownLatch(1); @@ -310,10 +313,10 @@ public CallbackResponse cursorReady(AsyncResultSet resultSet) { assertThat(get(finished)).isNull(); ApiFuture commit = connection.commitAsync(); - assertThat(get(rowCount)).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); - assertThat(get(updateCount)).isEqualTo(UPDATE_COUNT); + assertEquals(RANDOM_RESULT_SET_ROW_COUNT, get(rowCount)); + assertEquals(UPDATE_COUNT, get(updateCount)); assertThat(get(commit)).isNull(); - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(1, counter.retryCount); // Verify the order of the statements on the server. List requests = @@ -322,16 +325,16 @@ public CallbackResponse cursorReady(AsyncResultSet resultSet) { mockSpanner.getRequests(), input -> input instanceof ExecuteSqlRequest)); // The entire transaction should be retried. assertThat(requests).hasSize(4); - assertThat(((ExecuteSqlRequest) requests.get(0)).getSeqno()).isEqualTo(1L); + assertEquals(1L, ((ExecuteSqlRequest) requests.get(0)).getSeqno()); assertThat(((ExecuteSqlRequest) requests.get(0)).getSql()) .isEqualTo(SELECT_RANDOM_STATEMENT.getSql()); - assertThat(((ExecuteSqlRequest) requests.get(1)).getSeqno()).isEqualTo(2L); + assertEquals(2L, ((ExecuteSqlRequest) requests.get(1)).getSeqno()); assertThat(((ExecuteSqlRequest) requests.get(1)).getSql()) .isEqualTo(INSERT_STATEMENT.getSql()); - assertThat(((ExecuteSqlRequest) requests.get(2)).getSeqno()).isEqualTo(1L); + assertEquals(1L, ((ExecuteSqlRequest) requests.get(2)).getSeqno()); assertThat(((ExecuteSqlRequest) requests.get(2)).getSql()) .isEqualTo(SELECT_RANDOM_STATEMENT.getSql()); - assertThat(((ExecuteSqlRequest) requests.get(3)).getSeqno()).isEqualTo(2L); + assertEquals(2L, ((ExecuteSqlRequest) requests.get(3)).getSeqno()); assertThat(((ExecuteSqlRequest) requests.get(3)).getSql()) .isEqualTo(INSERT_STATEMENT.getSql()); } @@ -345,7 +348,7 @@ public void testUpdateAndQueryAbortedMidway_UpdateCountChanged() throws Interrup RANDOM_RESULT_SET_ROW_COUNT / 2)); final RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertThat(counter.retryCount).isEqualTo(0); + assertEquals(0, counter.retryCount); final CountDownLatch updateLatch = new CountDownLatch(1); final CountDownLatch queryLatch = new CountDownLatch(1); ApiFuture finished; @@ -390,7 +393,7 @@ public void testUpdateAndQueryAbortedMidway_UpdateCountChanged() throws Interrup get(finished); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(1, counter.retryCount); } finally { mockSpanner.putStatementResult(StatementResult.update(INSERT_STATEMENT, UPDATE_COUNT)); } @@ -403,16 +406,16 @@ public void testUpdateAndQueryAbortedMidway_UpdateCountChanged() throws Interrup // The entire transaction should be retried, but will not succeed as the result of the update // statement was different during the retry. assertThat(requests).hasSize(4); - assertThat(((ExecuteSqlRequest) requests.get(0)).getSeqno()).isEqualTo(1L); + assertEquals(1L, ((ExecuteSqlRequest) requests.get(0)).getSeqno()); assertThat(((ExecuteSqlRequest) requests.get(0)).getSql()) .isEqualTo(SELECT_RANDOM_STATEMENT.getSql()); - assertThat(((ExecuteSqlRequest) requests.get(1)).getSeqno()).isEqualTo(2L); + assertEquals(2L, ((ExecuteSqlRequest) requests.get(1)).getSeqno()); assertThat(((ExecuteSqlRequest) requests.get(1)).getSql()) .isEqualTo(INSERT_STATEMENT.getSql()); - assertThat(((ExecuteSqlRequest) requests.get(2)).getSeqno()).isEqualTo(1L); + assertEquals(1L, ((ExecuteSqlRequest) requests.get(2)).getSeqno()); assertThat(((ExecuteSqlRequest) requests.get(2)).getSql()) .isEqualTo(SELECT_RANDOM_STATEMENT.getSql()); - assertThat(((ExecuteSqlRequest) requests.get(3)).getSeqno()).isEqualTo(2L); + assertEquals(2L, ((ExecuteSqlRequest) requests.get(3)).getSeqno()); assertThat(((ExecuteSqlRequest) requests.get(3)).getSql()) .isEqualTo(INSERT_STATEMENT.getSql()); } @@ -488,7 +491,7 @@ public CallbackResponse cursorReady(AsyncResultSet resultSet) { get(res1); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(1, counter.retryCount); } } } @@ -501,8 +504,8 @@ public void testBlindUpdateAborted() { ApiFuture updateCount = connection.executeUpdateAsync(INSERT_STATEMENT); get(connection.commitAsync()); - assertThat(get(updateCount)).isEqualTo(UPDATE_COUNT); - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(UPDATE_COUNT, get(updateCount)); + assertEquals(1, counter.retryCount); } } @@ -526,7 +529,7 @@ public void testBlindUpdateAborted_WithConcurrentModification() { get(connection.commitAsync()); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(1, counter.retryCount); } } } @@ -548,7 +551,7 @@ public void testCommitAfterBlindUpdateAborted_WithConcurrentModification() { get(insert); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(1, counter.retryCount); } // Committing a transaction that has been aborted must fail and must not silently succeed. @@ -556,7 +559,7 @@ public void testCommitAfterBlindUpdateAborted_WithConcurrentModification() { get(connection.commitAsync()); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(1, counter.retryCount); } } } @@ -588,7 +591,7 @@ public void testMultipleBlindUpdatesAborted_WithConcurrentModification() { get(fut); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(1, counter.retryCount); } } } @@ -625,7 +628,7 @@ public void testBlindUpdateAborted_ThenAsyncQuery_WithConcurrentModification() { assertThat(get(fut)).isNull(); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(1, counter.retryCount); } } @@ -670,7 +673,7 @@ public void testBlindUpdateAborted_SelectResults() { get(commit); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertThat(counter.retryCount).isEqualTo(1); + assertEquals(1, counter.retryCount); } } } From ce700b4dbfd59dd6303312a0f91345d14049fc09 Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Tue, 27 Oct 2026 15:42:28 +0000 Subject: [PATCH 3/6] Addressed comments --- .../ConnectionAsyncApiAbortedTest.java | 90 +++++++++---------- 1 file changed, 45 insertions(+), 45 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java index 147a3d9f8c60..a65b06a30582 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java @@ -141,15 +141,15 @@ protected Builder configureConnectionOptions(Builder builder) { public void testSingleQueryAborted() { RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertEquals(0, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(0); mockSpanner.setExecuteStreamingSqlExecutionTime( SimulatedExecutionTime.ofException( mockSpanner.createAbortedException(ByteString.copyFromUtf8("test")))); QueryResult res = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT); assertThat(get(res.finished)).isNull(); - assertEquals(RANDOM_RESULT_SET_ROW_COUNT, res.rowCount.get()); - assertEquals(1, counter.retryCount); + assertThat(res.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); + assertThat(counter.retryCount).isEqualTo(1); } } @@ -157,7 +157,7 @@ public void testSingleQueryAborted() { public void testTwoQueriesSecondAborted() { RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertEquals(0, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(0); QueryResult res1 = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT); mockSpanner.setExecuteStreamingSqlExecutionTime( SimulatedExecutionTime.ofException( @@ -165,10 +165,10 @@ public void testTwoQueriesSecondAborted() { QueryResult res2 = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT_2); assertThat(get(res1.finished)).isNull(); - assertEquals(RANDOM_RESULT_SET_ROW_COUNT, res1.rowCount.get()); + assertThat(res1.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); assertThat(get(res2.finished)).isNull(); - assertEquals(RANDOM_RESULT_SET_ROW_COUNT_2, res2.rowCount.get()); - assertEquals(1, counter.retryCount); + assertThat(res2.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT_2); + assertThat(counter.retryCount).isEqualTo(1); } } @@ -176,7 +176,7 @@ public void testTwoQueriesSecondAborted() { public void testTwoQueriesBothAborted() throws InterruptedException { RetryCounter counter = new RetryCounter(1); try (Connection connection = createConnection(counter)) { - assertEquals(0, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(0); mockSpanner.setExecuteStreamingSqlExecutionTime( SimulatedExecutionTime.ofException( mockSpanner.createAbortedException(ByteString.copyFromUtf8("test")))); @@ -189,10 +189,10 @@ public void testTwoQueriesBothAborted() throws InterruptedException { QueryResult res2 = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT_2); assertThat(get(res1.finished)).isNull(); - assertEquals(RANDOM_RESULT_SET_ROW_COUNT, res1.rowCount.get()); + assertThat(res1.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); assertThat(get(res2.finished)).isNull(); - assertEquals(RANDOM_RESULT_SET_ROW_COUNT_2, res2.rowCount.get()); - assertEquals(2, counter.retryCount); + assertThat(res2.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT_2); + assertThat(counter.retryCount).isEqualTo(2); } } @@ -204,12 +204,12 @@ public void testSingleQueryAbortedMidway() { RANDOM_RESULT_SET_ROW_COUNT / 2)); RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertEquals(0, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(0); QueryResult res = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT); assertThat(get(res.finished)).isNull(); - assertEquals(RANDOM_RESULT_SET_ROW_COUNT, res.rowCount.get()); - assertEquals(1, counter.retryCount); + assertThat(res.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); + assertThat(counter.retryCount).isEqualTo(1); } } @@ -217,7 +217,7 @@ public void testSingleQueryAbortedMidway() { public void testTwoQueriesSecondAbortedMidway() { RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertEquals(0, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(0); QueryResult res1 = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT); mockSpanner.setExecuteStreamingSqlExecutionTime( SimulatedExecutionTime.ofStreamException( @@ -226,10 +226,10 @@ public void testTwoQueriesSecondAbortedMidway() { QueryResult res2 = executeQueryAsync(connection, SELECT_RANDOM_STATEMENT_2); assertThat(get(res1.finished)).isNull(); - assertEquals(RANDOM_RESULT_SET_ROW_COUNT, res1.rowCount.get()); + assertThat(res1.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); assertThat(get(res2.finished)).isNull(); - assertEquals(RANDOM_RESULT_SET_ROW_COUNT_2, res2.rowCount.get()); - assertEquals(1, counter.retryCount); + assertThat(res2.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT_2); + assertThat(counter.retryCount).isEqualTo(1); } } @@ -241,7 +241,7 @@ public void testTwoQueriesOneAbortedMidway() { Math.min(RANDOM_RESULT_SET_ROW_COUNT / 2, RANDOM_RESULT_SET_ROW_COUNT_2 / 2))); RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertEquals(0, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(0); // These AsyncResultSets will be consumed in parallel. One of them will (at random) abort // halfway. QueryResult res1 = @@ -250,10 +250,10 @@ public void testTwoQueriesOneAbortedMidway() { executeQueryAsync(connection, SELECT_RANDOM_STATEMENT_2, multiThreadedExecutor); assertThat(get(res1.finished)).isNull(); - assertEquals(RANDOM_RESULT_SET_ROW_COUNT, res1.rowCount.get()); + assertThat(res1.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); assertThat(get(res2.finished)).isNull(); - assertEquals(RANDOM_RESULT_SET_ROW_COUNT_2, res2.rowCount.get()); - assertEquals(1, counter.retryCount); + assertThat(res2.rowCount.get()).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT_2); + assertThat(counter.retryCount).isEqualTo(1); } } @@ -265,7 +265,7 @@ public void testUpdateAndQueryAbortedMidway() throws InterruptedException { RANDOM_RESULT_SET_ROW_COUNT / 2)); final RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertEquals(0, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(0); final SettableApiFuture rowCount = SettableApiFuture.create(); final CountDownLatch updateLatch = new CountDownLatch(1); final CountDownLatch queryLatch = new CountDownLatch(1); @@ -313,10 +313,10 @@ public CallbackResponse cursorReady(AsyncResultSet resultSet) { assertThat(get(finished)).isNull(); ApiFuture commit = connection.commitAsync(); - assertEquals(RANDOM_RESULT_SET_ROW_COUNT, get(rowCount)); - assertEquals(UPDATE_COUNT, get(updateCount)); + assertThat(get(rowCount)).isEqualTo(RANDOM_RESULT_SET_ROW_COUNT); + assertThat(get(updateCount)).isEqualTo(UPDATE_COUNT); assertThat(get(commit)).isNull(); - assertEquals(1, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(1); // Verify the order of the statements on the server. List requests = @@ -325,16 +325,16 @@ public CallbackResponse cursorReady(AsyncResultSet resultSet) { mockSpanner.getRequests(), input -> input instanceof ExecuteSqlRequest)); // The entire transaction should be retried. assertThat(requests).hasSize(4); - assertEquals(1L, ((ExecuteSqlRequest) requests.get(0)).getSeqno()); + assertThat(((ExecuteSqlRequest) requests.get(0)).getSeqno()).isEqualTo(1L); assertThat(((ExecuteSqlRequest) requests.get(0)).getSql()) .isEqualTo(SELECT_RANDOM_STATEMENT.getSql()); - assertEquals(2L, ((ExecuteSqlRequest) requests.get(1)).getSeqno()); + assertThat(((ExecuteSqlRequest) requests.get(1)).getSeqno()).isEqualTo(2L); assertThat(((ExecuteSqlRequest) requests.get(1)).getSql()) .isEqualTo(INSERT_STATEMENT.getSql()); - assertEquals(1L, ((ExecuteSqlRequest) requests.get(2)).getSeqno()); + assertThat(((ExecuteSqlRequest) requests.get(2)).getSeqno()).isEqualTo(1L); assertThat(((ExecuteSqlRequest) requests.get(2)).getSql()) .isEqualTo(SELECT_RANDOM_STATEMENT.getSql()); - assertEquals(2L, ((ExecuteSqlRequest) requests.get(3)).getSeqno()); + assertThat(((ExecuteSqlRequest) requests.get(3)).getSeqno()).isEqualTo(2L); assertThat(((ExecuteSqlRequest) requests.get(3)).getSql()) .isEqualTo(INSERT_STATEMENT.getSql()); } @@ -348,7 +348,7 @@ public void testUpdateAndQueryAbortedMidway_UpdateCountChanged() throws Interrup RANDOM_RESULT_SET_ROW_COUNT / 2)); final RetryCounter counter = new RetryCounter(); try (Connection connection = createConnection(counter)) { - assertEquals(0, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(0); final CountDownLatch updateLatch = new CountDownLatch(1); final CountDownLatch queryLatch = new CountDownLatch(1); ApiFuture finished; @@ -393,7 +393,7 @@ public void testUpdateAndQueryAbortedMidway_UpdateCountChanged() throws Interrup get(finished); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertEquals(1, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(1); } finally { mockSpanner.putStatementResult(StatementResult.update(INSERT_STATEMENT, UPDATE_COUNT)); } @@ -406,16 +406,16 @@ public void testUpdateAndQueryAbortedMidway_UpdateCountChanged() throws Interrup // The entire transaction should be retried, but will not succeed as the result of the update // statement was different during the retry. assertThat(requests).hasSize(4); - assertEquals(1L, ((ExecuteSqlRequest) requests.get(0)).getSeqno()); + assertThat(((ExecuteSqlRequest) requests.get(0)).getSeqno()).isEqualTo(1L); assertThat(((ExecuteSqlRequest) requests.get(0)).getSql()) .isEqualTo(SELECT_RANDOM_STATEMENT.getSql()); - assertEquals(2L, ((ExecuteSqlRequest) requests.get(1)).getSeqno()); + assertThat(((ExecuteSqlRequest) requests.get(1)).getSeqno()).isEqualTo(2L); assertThat(((ExecuteSqlRequest) requests.get(1)).getSql()) .isEqualTo(INSERT_STATEMENT.getSql()); - assertEquals(1L, ((ExecuteSqlRequest) requests.get(2)).getSeqno()); + assertThat(((ExecuteSqlRequest) requests.get(2)).getSeqno()).isEqualTo(1L); assertThat(((ExecuteSqlRequest) requests.get(2)).getSql()) .isEqualTo(SELECT_RANDOM_STATEMENT.getSql()); - assertEquals(2L, ((ExecuteSqlRequest) requests.get(3)).getSeqno()); + assertThat(((ExecuteSqlRequest) requests.get(3)).getSeqno()).isEqualTo(2L); assertThat(((ExecuteSqlRequest) requests.get(3)).getSql()) .isEqualTo(INSERT_STATEMENT.getSql()); } @@ -491,7 +491,7 @@ public CallbackResponse cursorReady(AsyncResultSet resultSet) { get(res1); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertEquals(1, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(1); } } } @@ -504,8 +504,8 @@ public void testBlindUpdateAborted() { ApiFuture updateCount = connection.executeUpdateAsync(INSERT_STATEMENT); get(connection.commitAsync()); - assertEquals(UPDATE_COUNT, get(updateCount)); - assertEquals(1, counter.retryCount); + assertThat(get(updateCount)).isEqualTo(UPDATE_COUNT); + assertThat(counter.retryCount).isEqualTo(1); } } @@ -529,7 +529,7 @@ public void testBlindUpdateAborted_WithConcurrentModification() { get(connection.commitAsync()); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertEquals(1, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(1); } } } @@ -551,7 +551,7 @@ public void testCommitAfterBlindUpdateAborted_WithConcurrentModification() { get(insert); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertEquals(1, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(1); } // Committing a transaction that has been aborted must fail and must not silently succeed. @@ -559,7 +559,7 @@ public void testCommitAfterBlindUpdateAborted_WithConcurrentModification() { get(connection.commitAsync()); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertEquals(1, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(1); } } } @@ -591,7 +591,7 @@ public void testMultipleBlindUpdatesAborted_WithConcurrentModification() { get(fut); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertEquals(1, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(1); } } } @@ -628,7 +628,7 @@ public void testBlindUpdateAborted_ThenAsyncQuery_WithConcurrentModification() { assertThat(get(fut)).isNull(); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertEquals(1, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(1); } } @@ -673,7 +673,7 @@ public void testBlindUpdateAborted_SelectResults() { get(commit); fail("Missing expected exception"); } catch (AbortedDueToConcurrentModificationException e) { - assertEquals(1, counter.retryCount); + assertThat(counter.retryCount).isEqualTo(1); } } } From 85166c705bb929434e910f4f902fc5988b44762c Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Tue, 27 Oct 2026 15:55:07 +0000 Subject: [PATCH 4/6] Addressed comments --- .../ConnectionAsyncApiAbortedTest.java | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java index a65b06a30582..1f57d476a076 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java @@ -547,20 +547,14 @@ public void testCommitAfterBlindUpdateAborted_WithConcurrentModification() { mockSpanner.abortNextStatement(); ApiFuture insert = connection.executeUpdateAsync(INSERT_STATEMENT); // Wait until the failed retry has marked the transaction as aborted before calling commit. - try { - get(insert); - fail("Missing expected exception"); - } catch (AbortedDueToConcurrentModificationException e) { - assertThat(counter.retryCount).isEqualTo(1); - } + assertThrows( + AbortedDueToConcurrentModificationException.class, () -> get(insert)); + assertEquals(1, counter.retryCount); // Committing a transaction that has been aborted must fail and must not silently succeed. - try { - get(connection.commitAsync()); - fail("Missing expected exception"); - } catch (AbortedDueToConcurrentModificationException e) { - assertThat(counter.retryCount).isEqualTo(1); - } + assertThrows( + AbortedDueToConcurrentModificationException.class, () -> get(connection.commitAsync())); + assertEquals(1, counter.retryCount); } } From 181d8083cedd6e0073f1d572c8978f30738a03bf Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Thu, 29 Oct 2026 09:53:49 +0000 Subject: [PATCH 5/6] chore(spanner): fix formatting --- .../cloud/spanner/connection/ReadWriteTransaction.java | 4 +++- .../spanner/connection/ConnectionAsyncApiAbortedTest.java | 6 ++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java index bd490e60296b..1e81a436ac58 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java @@ -1329,7 +1329,9 @@ private void invokeTransactionRetryListenersOnFinish(RetryResult result) { @Override public Void call() { try { - if (state != UnitOfWorkState.ABORTED && abortedException == null && rolledBackToSavepointException == null) { + if (state != UnitOfWorkState.ABORTED + && abortedException == null + && rolledBackToSavepointException == null) { // Make sure the transaction has actually started before we try to rollback. get(txContextFuture); txManager.rollback(); diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java index 1f57d476a076..919094042a0a 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java @@ -18,10 +18,9 @@ import static com.google.cloud.spanner.SpannerApiFutures.get; import static com.google.common.truth.Truth.assertThat; -import static org.junit.Assert.fail; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertThrows; - +import static org.junit.Assert.fail; import com.google.api.core.ApiFuture; import com.google.api.core.SettableApiFuture; @@ -547,8 +546,7 @@ public void testCommitAfterBlindUpdateAborted_WithConcurrentModification() { mockSpanner.abortNextStatement(); ApiFuture insert = connection.executeUpdateAsync(INSERT_STATEMENT); // Wait until the failed retry has marked the transaction as aborted before calling commit. - assertThrows( - AbortedDueToConcurrentModificationException.class, () -> get(insert)); + assertThrows(AbortedDueToConcurrentModificationException.class, () -> get(insert)); assertEquals(1, counter.retryCount); // Committing a transaction that has been aborted must fail and must not silently succeed. From 47972e68df48efa5db5f01c46534cb261d5acb7c Mon Sep 17 00:00:00 2001 From: Sakthivel Subramanian Date: Fri, 30 Oct 2026 10:08:25 +0000 Subject: [PATCH 6/6] Addressed comments --- .../spanner/connection/ConnectionImpl.java | 12 +-- .../connection/ReadWriteTransaction.java | 24 +++-- .../ConnectionAsyncApiAbortedTest.java | 101 +++++++++++++++++- 3 files changed, 118 insertions(+), 19 deletions(-) diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionImpl.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionImpl.java index 3e191023c3b5..a1ed225c71cc 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionImpl.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ConnectionImpl.java @@ -464,7 +464,7 @@ public ApiFuture closeAsync() { if (isBatchActive()) { abortBatch(); } - if (isTransactionStarted()) { + if (isTransactionStarted() || isCurrentTransactionAborted()) { try { futures.add(rollbackAsync()); } catch (Exception exception) { @@ -1143,9 +1143,10 @@ private boolean hasTransactionalChanges() { } /** - * Returns true if the current transaction has been aborted and could not be retried. Committing - * such a transaction must return the abort error instead of silently succeeding. This can happen - * when a statement that was executed asynchronously aborts before commit is called. + * Returns true if the current transaction has been aborted and could not be retried. Ending such + * a transaction must be delegated to the underlying unit of work so that committing returns the + * abort error instead of silently succeeding, and rolling back transitions the state to + * ROLLED_BACK, cancels any scheduled keep-alive ping, and ends the OpenTelemetry span. */ private boolean isCurrentTransactionAborted() { return internalIsInTransaction() @@ -1398,8 +1399,7 @@ private ApiFuture endCurrentTransactionAsync( statementTag == null, "Statement tags are not supported for COMMIT or ROLLBACK"); ApiFuture res; try { - if (hasTransactionalChanges() - || (endTransactionMethod == commit && isCurrentTransactionAborted())) { + if (hasTransactionalChanges() || isCurrentTransactionAborted()) { res = endTransactionMethod.endAsync( callType, getCurrentUnitOfWorkOrStartNewUnitOfWork(parsedStatement)); diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java index 1e81a436ac58..594e8c6f6dde 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ReadWriteTransaction.java @@ -473,7 +473,8 @@ private void checkValidStateAndMarkStarted() { } private boolean shouldPing() { - return isActive() + return getState().isActive() + && this.abortedException == null && keepAliveLock != null && keepTransactionAlive && !timedOutOrCancelled @@ -1245,8 +1246,7 @@ private void handleAborted(AbortedException aborted) { } catch (Throwable t) { // ignore } - this.state = UnitOfWorkState.ABORTED; - this.abortedException = e; + markAborted(e); throw e; } catch (AbortedException abortedExceptionDuringRetry) { // Retry aborted, do another retry of the transaction. @@ -1271,8 +1271,7 @@ private void handleAborted(AbortedException aborted) { // ignore } // Set transaction state to aborted as the retry failed. - this.state = UnitOfWorkState.ABORTED; - this.abortedException = aborted; + markAborted(aborted); // Re-throw underlying exception. throw e; } @@ -1284,12 +1283,18 @@ private void handleAborted(AbortedException aborted) { // ignore } // Internal retry is not enabled. - this.state = UnitOfWorkState.ABORTED; - this.abortedException = aborted; + markAborted(aborted); throw aborted; } } + private void markAborted(AbortedException aborted) { + if (this.state == UnitOfWorkState.STARTED) { + this.state = UnitOfWorkState.ABORTED; + } + this.abortedException = aborted; + } + private void throwAbortWithRetryAttemptsExceeded() throws SpannerException { invokeTransactionRetryListenersOnFinish(RetryResult.RETRY_ABORTED_AND_MAX_ATTEMPTS_EXCEEDED); logger.fine( @@ -1304,11 +1309,10 @@ private void throwAbortWithRetryAttemptsExceeded() throws SpannerException { } catch (Throwable t) { // ignore } - this.state = UnitOfWorkState.ABORTED; - this.abortedException = + markAborted( (AbortedException) SpannerExceptionFactory.newSpannerException( - ErrorCode.ABORTED, MAX_INTERNAL_RETRIES_EXCEEDED); + ErrorCode.ABORTED, MAX_INTERNAL_RETRIES_EXCEEDED)); throw this.abortedException; } diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java index 919094042a0a..93512fe4b9dc 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ConnectionAsyncApiAbortedTest.java @@ -19,6 +19,9 @@ import static com.google.cloud.spanner.SpannerApiFutures.get; import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; import static org.junit.Assert.fail; @@ -37,6 +40,7 @@ import com.google.cloud.spanner.connection.ConnectionOptions.Builder; import com.google.cloud.spanner.connection.ITAbstractSpannerTest.ITConnection; import com.google.cloud.spanner.connection.StatementExecutor.StatementExecutorType; +import com.google.cloud.spanner.connection.UnitOfWork.UnitOfWorkState; import com.google.common.collect.Collections2; import com.google.common.collect.ImmutableList; import com.google.common.collect.Lists; @@ -72,6 +76,7 @@ private static final class QueryResult { private static final class RetryCounter implements TransactionRetryListener { final CountDownLatch latch; + final CountDownLatch finishedLatch; int retryCount = 0; RetryCounter() { @@ -79,7 +84,12 @@ private static final class RetryCounter implements TransactionRetryListener { } RetryCounter(int countDown) { - latch = new CountDownLatch(countDown); + this(countDown, 0); + } + + RetryCounter(int startCountDown, int finishCountDown) { + latch = new CountDownLatch(startCountDown); + finishedLatch = new CountDownLatch(finishCountDown); } @Override @@ -90,7 +100,9 @@ public void retryStarting(Timestamp transactionStarted, long transactionId, int @Override public void retryFinished( - Timestamp transactionStarted, long transactionId, int retryAttempt, RetryResult result) {} + Timestamp transactionStarted, long transactionId, int retryAttempt, RetryResult result) { + finishedLatch.countDown(); + } } private static final ExecutorService singleThreadedExecutor = Executors.newSingleThreadExecutor(); @@ -534,7 +546,7 @@ public void testBlindUpdateAborted_WithConcurrentModification() { } @Test - public void testCommitAfterBlindUpdateAborted_WithConcurrentModification() { + public void testCommitAfterUpdateAborted_WithConcurrentModification() { Statement update1 = Statement.of("UPDATE FOO SET BAR=1 WHERE BAZ=100"); mockSpanner.putStatementResult(StatementResult.update(update1, 100)); @@ -556,6 +568,89 @@ public void testCommitAfterBlindUpdateAborted_WithConcurrentModification() { } } + @Test + public void testBlindUpdateAborted_ConcurrentCommit_WithConcurrentModification() + throws InterruptedException { + Statement update1 = Statement.of("UPDATE FOO SET BAR=1 WHERE BAZ=100"); + for (int i = 0; i < 10; i++) { + mockSpanner.putStatementResult(StatementResult.update(update1, 100)); + + RetryCounter counter = new RetryCounter(0, 1); + try (Connection connection = createConnection(counter)) { + get(connection.executeUpdateAsync(update1)); + mockSpanner.putStatementResult(StatementResult.update(update1, 200)); + + mockSpanner.abortNextStatement(); + // Execute a blind (shoot-and-forget) update without waiting for its future. + connection.executeUpdateAsync(INSERT_STATEMENT); + // Trigger commitAsync concurrently as the internal retry finishes (while the blind update + // is still rolling back and marking the transaction as aborted on the executor thread). + assertThat(counter.finishedLatch.await(10L, TimeUnit.SECONDS)).isTrue(); + assertThrows( + AbortedDueToConcurrentModificationException.class, () -> get(connection.commitAsync())); + assertEquals(1, counter.retryCount); + } + } + } + + @Test + public void testRollbackAfterTransactionAborted_UpdatesStateToRolledBack() { + Statement update1 = Statement.of("UPDATE FOO SET BAR=1 WHERE BAZ=100"); + mockSpanner.putStatementResult(StatementResult.update(update1, 100)); + + RetryCounter counter = new RetryCounter(); + try (Connection connection = createConnection(counter)) { + get(connection.executeUpdateAsync(update1)); + mockSpanner.putStatementResult(StatementResult.update(update1, 200)); + + mockSpanner.abortNextStatement(); + ApiFuture insert = connection.executeUpdateAsync(INSERT_STATEMENT); + assertThrows(AbortedDueToConcurrentModificationException.class, () -> get(insert)); + assertEquals(1, counter.retryCount); + + // Verify the transaction is currently in the ABORTED state. + ConnectionImpl connectionImpl = (ConnectionImpl) connection; + ReadWriteTransaction transaction = + (ReadWriteTransaction) connectionImpl.getCurrentUnitOfWorkOrStartNewUnitOfWork(); + assertEquals(UnitOfWorkState.ABORTED, transaction.getState()); + + // Rollback the aborted transaction. + get(connection.rollbackAsync()); + + assertEquals(UnitOfWorkState.ROLLED_BACK, transaction.getState()); + } + } + + @Test + public void testRollbackAfterTransactionAborted_CancelsKeepAlivePing() { + Statement update1 = Statement.of("UPDATE FOO SET BAR=1 WHERE BAZ=100"); + mockSpanner.putStatementResult(StatementResult.update(update1, 100)); + + RetryCounter counter = new RetryCounter(); + try (Connection connection = createConnection(counter)) { + connection.setKeepTransactionAlive(true); + get(connection.executeUpdateAsync(update1)); + mockSpanner.putStatementResult(StatementResult.update(update1, 200)); + + ConnectionImpl connectionImpl = (ConnectionImpl) connection; + ReadWriteTransaction transaction = + (ReadWriteTransaction) connectionImpl.getCurrentUnitOfWorkOrStartNewUnitOfWork(); + assertNotNull(transaction.getKeepAliveFuture()); + assertFalse(transaction.getKeepAliveFuture().isCancelled()); + + mockSpanner.abortNextStatement(); + ApiFuture insert = connection.executeUpdateAsync(INSERT_STATEMENT); + assertThrows(AbortedDueToConcurrentModificationException.class, () -> get(insert)); + assertEquals(1, counter.retryCount); + assertEquals(UnitOfWorkState.ABORTED, transaction.getState()); + + // Rollback the aborted transaction. + get(connection.rollbackAsync()); + + assertNull(transaction.getKeepAliveFuture()); + } + } + @Test public void testMultipleBlindUpdatesAborted_WithConcurrentModification() { Statement update1 = Statement.of("UPDATE FOO SET BAR=1 WHERE BAZ=100");