chore: add explanation + test for aborted tx

Author: olavloite

cloud-spanner-r2dbc/src/main/java/com/google/cloud/spanner/r2dbc/v2/DatabaseClientTransactionManager.java

@@ -109,6 +109,10 @@ ApiFuture<Void> clearTransactionManager() {
     ApiFuture<Void> returnFuture = ApiFutures.immediateFuture(null);
 
     if (this.transactionManager != null) {
+      // We don't close the transaction manager here if it has already been rolled back.
+      // Rolling back a transaction manager automatically also closes it, and returns the underlying
+      // session to the pool. Closing it a second time here, would cause it to be added to the pool
+      // a second time.
       if (this.transactionManager.getState() != TransactionState.ROLLED_BACK) {
         returnFuture = this.transactionManager.closeAsync();
       }

cloud-spanner-r2dbc/src/test/java/com/google/cloud/spanner/r2dbc/v2/MockServerTest.java

@@ -20,9 +20,11 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.google.cloud.NoCredentials;
+import com.google.cloud.spanner.AbortedException;
 import com.google.cloud.spanner.MockSpannerServiceImpl;
 import com.google.cloud.spanner.MockSpannerServiceImpl.StatementResult;
 import com.google.cloud.spanner.Statement;
@@ -242,6 +244,29 @@ public void testRollbackTransaction() {
     assertEquals(0, mockSpanner.countRequestsOfType(CommitRequest.class));
   }
 
+  @Test
+  public void testAbortedTransaction() {
+    String sql = "insert into foo (id) values (1)";
+    mockSpanner.putStatementResult(StatementResult.update(Statement.of(sql), 1L));
+
+    ConnectionFactory connectionFactory = createConnectionFactory();
+    Publisher<? extends Connection> connectionPublisher = connectionFactory.create();
+    Connection connection = Mono.from(connectionPublisher).block();
+
+    mockSpanner.abortNextTransaction();
+    assertThrows(AbortedException.class, () -> Flux.concat(
+        connection.beginTransaction(),
+        connection.createStatement(sql).execute(),
+        connection.commitTransaction()
+    ).blockLast());
+
+    Mono.from(((Closeable) connectionFactory).close()).subscribe();
+
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class));
+    assertEquals(0, mockSpanner.countRequestsOfType(RollbackRequest.class));
+    assertEquals(1, mockSpanner.countRequestsOfType(CommitRequest.class));
+  }
+
   @Test
   public void testRollbackEmptyTransactionAndExecuteSqlInParallel() throws Exception {
     // Redirect log to /dev/null.