AsyncTransactionManager needs to roll back active transaction when closed

[elefeint]

When testing the new R2DBC driver with the Cloud Spanner emulator, I noticed that the active transaction was not being rolled back despite AsyncTransactionManager.close() being called.

The contract for close() is to roll back the active transaction when closed, but I wonder if close() was not meant to be implemented as closeAsync() and kind of got left out.

Environment details

Spanner client library version: 1.59.0

Steps to reproduce

1. Start emulator and set the emulator environment variable (SPANNER_EMULATOR_HOST=localhost:9010). The issue has nothing to do with the emulator, but it's really easy to observe when using one.
2. Update "project", "instance", "database" to real values.
3. Set up the following table definition:
   gcloud spanner databases ddl update database --ddl="CREATE TABLE test ( value INT64 ) PRIMARY KEY (value)" --instance=instance
4. Run transactionManagerRollsBackTransactionWhenClosed twice; observe that everything works as expected.
5. Run asyncTransactionManagerDoesNotRollBackTransactionWhenClosed twice; observe that the first run succeeds, but the second fails with error io.grpc.StatusRuntimeException: ABORTED: Transaction NNN aborted due to active transaction MMM. The emulator only supports one transaction at a time.

  DatabaseId databaseId = DatabaseId.of("project", "instance", "database");
  SpannerOptions options = SpannerOptions.newBuilder().build();
  Spanner spanner = options.getService();
  DatabaseClient dbClient = null;
  TransactionManager txnManager = null;
  AsyncTransactionManager asyncTxnManager = null;

  @Test
  public void transactionManagerRollsBackTransactionWhenClosed() throws Exception {

    try {
      dbClient = spanner.getDatabaseClient(databaseId);
      txnManager = dbClient.transactionManager();
      TransactionContext ctx = txnManager.begin();
      ctx.executeUpdate(Statement.newBuilder("INSERT INTO test (value) VALUES (1234567)").build());

      // MISSING COMMIT; expecting rollback
      // tm.commit();

    } finally {

      if (txnManager != null) {
        System.out.println("************ CLOSING TRANSACTION MANAGER");
        txnManager.close();
      }
    }
  }

  @Test
  public void asyncTransactionManagerDoesNotRollBackTransactionWhenClosed() throws Exception {

    try {
      dbClient = spanner.getDatabaseClient(databaseId);
      asyncTxnManager = dbClient.transactionManagerAsync();
      TransactionContext ctx = asyncTxnManager.beginAsync().get();
      ctx.executeUpdateAsync(Statement.newBuilder("INSERT INTO test (value) VALUES (1234567)").build())
        .get();

      // MISSING COMMIT; expecting rollback upon transaction manager closing
      // tm.commit();

    } finally {

      if (asyncTxnManager != null) {
        asyncTxnManager.close();
      }
    }
  }

[olavloite]

@elefeint Thanks for the detailed report and easy to reproduce example. I've provided a fix for it in #505.