fix: AsyncTransactionManager did not always close the session (#3580)

In a case where the following happens, the session that was used by an AsyncTransactionManager
would not be returned to the pool:
1. The transaction is abandoned without a Commit or Rollback call.
2. The AsyncTransactionManager then executes a RollbackAsync internally.
3. If the internal RollbackAsync failed, then the session would not be closed.

google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionPoolAsyncTransactionManager.java

@@ -107,6 +107,7 @@ public void onSuccess(AsyncTransactionManagerImpl result) {
                 new ApiFutureCallback<Void>() {
                   @Override
                   public void onFailure(Throwable t) {
+                    session.close();
                     res.setException(t);
                   }
 

google-cloud-spanner/src/test/java/com/google/cloud/spanner/AsyncTransactionManagerTest.java

@@ -27,6 +27,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutureCallback;
@@ -1084,6 +1085,37 @@ public void onSuccess(Long aLong) {
     }
   }
 
+  @Test
+  public void testAbandonedAsyncTransactionManager_rollbackFails() throws Exception {
+    mockSpanner.setRollbackExecutionTime(
+        SimulatedExecutionTime.ofException(Status.PERMISSION_DENIED.asRuntimeException()));
+
+    boolean gotException = false;
+    try (AsyncTransactionManager manager = client().transactionManagerAsync()) {
+      TransactionContextFuture transactionContextFuture = manager.beginAsync();
+      while (true) {
+        try {
+          AsyncTransactionStep<Void, Long> updateCount =
+              transactionContextFuture.then(
+                  (transactionContext, ignored) ->
+                      transactionContext.executeUpdateAsync(UPDATE_STATEMENT),
+                  executor);
+          assertEquals(1L, updateCount.get().longValue());
+          // Break without committing or rolling back the transaction.
+          break;
+        } catch (AbortedException e) {
+          transactionContextFuture = manager.resetForRetryAsync();
+        }
+      }
+    } catch (SpannerException spannerException) {
+      // The error from the automatically executed Rollback is surfaced when the
+      // AsyncTransactionManager is closed.
+      assertEquals(ErrorCode.PERMISSION_DENIED, spannerException.getErrorCode());
+      gotException = true;
+    }
+    assertTrue(gotException);
+  }
+
   private boolean isMultiplexedSessionsEnabled() {
     if (spanner.getOptions() == null || spanner.getOptions().getSessionPoolOptions() == null) {
       return false;