Currently active transaction should be rolled back on connection close

[elefeint]

I was testing the emulator, and found a test bug in which a transaction was started, and then neither committed nor rolled back. However, connection was closed properly. Spanner emulator is very strict about leaving transactions hanging being a Bad Life Choice (you'll have to restart the emulator to get out of this scenario), which is how I realized there was a problem in the test I was working with.

If the connection is closed without an explicit commit/rollback, we should be calling rollback in the driver.

[elefeint]

Filed googleapis/java-spanner#504 to check AsynchronousTransactionManager.close() behavior.

[elefeint]

Re-test when google-cloud-spanner:2.0.3 is released.

[elefeint]

Re-tested; works great with the latest client library.

For posterity, the scenario is to run connection.beginTransaction() but not commit/rollback, instead following it with connection.close(). We no longer have this test in the codebase since the extended TCK syntax made overrides unnecessary.

Infrastructure steps:

1. Start emulator gcloud beta emulators spanner start
2. Switch to emulator config (see here if setting up for the first time): gcloud config configurations activate emulator
3. Create instance in emulator gcloud spanner instances create <INSTANCE> --config=emulator-config --description="Test Instance" --nodes=1
4. Create database gcloud spanner databases create --instance=<INSTANCE> <DATABASE>
5. Set up schema gcloud spanner databases ddl update <DATABASE> --ddl="CREATE TABLE test ( value INT64 ) PRIMARY KEY (value)" --instance=<INSTANCE>

Code (run twice to make sure bug is gone); set up environment variable SPANNER_EMULATOR_HOST=localhost:9010 before running:

  @Test
  public void bindNull() {
    Mono.from(getConnectionFactory().create())
        .delayUntil(c -> c.beginTransaction())
        .flatMapMany(connection -> Flux.from(connection

            .createStatement(
                String.format("INSERT INTO test (value) VALUES(%s)", getPlaceholder(0)))
            .bindNull(getIdentifier(0), Integer.class).add()
            .execute())

            .concatWith(close(connection)))
        .as(StepVerifier::create)
        .expectNextCount(1).as("rows inserted")
        .verifyComplete();
  }