fix(spanner): cancel the Commit RPC on commit timeout or interrupt

`TransactionContextImpl.commit()` cancels the in-flight Commit RPC when the
calling thread is interrupted or the bounded `get()` times out:

    } catch (InterruptedException | TimeoutException e) {
      if (commitFuture != null) {
        commitFuture.cancel(true);
      }

But the volatile field `commitFuture` was never assigned. The only
assignment done in `CommitRunnable.run()` was to a local variable that
shadowed it.

So the field stayed `null` and `commitFuture.cancel(true)` was dead code: on
interrupt/timeout the client abandoned the call while the Commit RPC could
still complete server-side.

Fix that by publishing the RPC future to the field, and name the field
`inFlightCommitFuture` for its cancellation role so it no longer collides with the
local. The `null` check still covers the case where commit times out before the
RPC is even sent.

Signed-off-by: Fredrik Fornwall <fredrik@fornwall.net>

Author: fornwall

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

@@ -365,8 +365,8 @@ void commit() {
                     rpc.getCommitRetrySettings().getTotalTimeout().getSeconds() + 5,
                     TimeUnit.SECONDS);
       } catch (InterruptedException | TimeoutException e) {
-        if (commitFuture != null) {
-          commitFuture.cancel(true);
+        if (inFlightCommitFuture != null) {
+          inFlightCommitFuture.cancel(true);
         }
         if (e instanceof InterruptedException) {
           throw SpannerExceptionFactory.propagateInterrupt((InterruptedException) e);
@@ -378,7 +378,7 @@ void commit() {
       }
     }
 
-    volatile ApiFuture<CommitResponse> commitFuture;
+    volatile ApiFuture<com.google.spanner.v1.CommitResponse> inFlightCommitFuture;
 
     ApiFuture<CommitResponse> commitAsync() {
       close();
@@ -491,6 +491,7 @@ public void run() {
           try (IScope ignore = tracer.withSpan(opSpan)) {
             commitFuture = rpc.commitAsync(commitRequest, getTransactionChannelHint());
           }
+          inFlightCommitFuture = commitFuture;
           session.markUsed(clock.instant());
           commitFuture.addListener(
               () -> {

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

@@ -20,6 +20,7 @@
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doThrow;
@@ -30,6 +31,7 @@
 import static org.mockito.Mockito.when;
 
 import com.google.api.core.ApiFutures;
+import com.google.api.core.SettableApiFuture;
 import com.google.cloud.grpc.GrpcTransportOptions;
 import com.google.cloud.grpc.GrpcTransportOptions.ExecutorFactory;
 import com.google.cloud.spanner.ErrorHandler.DefaultErrorHandler;
@@ -195,6 +197,36 @@ public void testCommitWithClientContext() {
     assertEquals(clientContext, capturedOptions.clientContext());
   }
 
+  @Test
+  public void commitCancelsInFlightRpcWhenCallingThreadInterrupted() {
+    when(session.getName()).thenReturn("projects/p/instances/i/databases/d/sessions/s");
+    TransactionContextImpl transaction =
+        TransactionContextImpl.newBuilder()
+            .setSession(session)
+            .setTransactionId(ByteString.copyFromUtf8("test-txn"))
+            .setOptions(Options.fromTransactionOptions())
+            .setRpc(rpc)
+            .setTracer(tracer)
+            .setSpan(span)
+            .build();
+    SettableApiFuture<CommitResponse> inFlightCommit = SettableApiFuture.create();
+    when(rpc.commitAsync(Mockito.any(CommitRequest.class), Mockito.anyMap()))
+        .thenAnswer(
+            invocation -> {
+              Thread.currentThread().interrupt();
+              return inFlightCommit;
+            });
+
+    try {
+      SpannerException e = assertThrows(SpannerException.class, transaction::commit);
+      assertEquals(ErrorCode.CANCELLED, e.getErrorCode());
+      assertTrue("in-flight Commit RPC was not cancelled", inFlightCommit.isCancelled());
+    } finally {
+      // Clear the interrupt flag so it cannot leak into other tests.
+      Thread.interrupted();
+    }
+  }
+
   @SuppressWarnings("unchecked")
   @Test
   public void usesPreparedTransaction() {