fix(spanner): prevent statement cancellation race in AbstractBaseUnitOfWork (#14283)

In `AbstractBaseUnitOfWork.executeStatementAsync`, statements were
submitted asynchronously to `statementExecutor` before storing the
resulting future in `currentlyRunningStatementFuture`. Under thread
contention, the query RPC can reach the mock server and trigger
`cancel()` before the future is assigned, causing `cancel()` to silently
drop the cancellation request and resulting in statement timeouts in CI
tests such as
`StatementTimeoutTest#testCancelReadWriteAutocommitMultipleStatements`.

Furthermore, submitting statements while holding the unit-of-work
monitor lock causes lock contention when
`StatementExecutorType.DIRECT_EXECUTOR` is used (such as in Spanner JDBC
or PGAdapter), because direct executors execute synchronously on the
calling thread.

This change:
- Pre-registers a `SettableApiFuture` as
`currentlyRunningStatementFuture` under synchronization before
submitting to the executor, ensuring any immediate `cancel()` invocation
is captured without racing.
- Submits the statement task to `statementExecutor` outside the
synchronized block to avoid holding the unit-of-work lock during direct
synchronous execution.
- Attaches callbacks to forward results/exceptions to the
`SettableApiFuture`, and registers a listener on the `SettableApiFuture`
to propagate cancellation to the underlying future and clear
`currentlyRunningStatementFuture`.

Author: sakthivelmanii

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/AbstractBaseUnitOfWork.java

@@ -17,7 +17,9 @@
 package com.google.cloud.spanner.connection;
 
 import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutureCallback;
 import com.google.api.core.ApiFutures;
+import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.longrunning.OperationFuture;
 import com.google.api.gax.rpc.ApiCallContext;
@@ -401,11 +403,29 @@ public <ReqT, RespT> ApiCallContext configure(
         io.opentelemetry.context.Context.current()
             .with(OpenTelemetryContextKeys.THREAD_NAME_KEY, Thread.currentThread().getName())
             .makeCurrent()) {
-      ApiFuture<T> f = statementExecutor.submit(context.wrap(callable));
       final SpannerAsyncExecutionException caller =
           callType == CallType.ASYNC
               ? new SpannerAsyncExecutionException(statement.getStatement())
               : null;
+      final SettableApiFuture<T> statementFuture = SettableApiFuture.create();
+      synchronized (this) {
+        this.currentlyRunningStatementFuture = statementFuture;
+      }
+      final ApiFuture<T> f;
+      try {
+        f = statementExecutor.submit(context.wrap(callable));
+      } catch (Throwable t) {
+        synchronized (this) {
+          if (this.currentlyRunningStatementFuture == statementFuture) {
+            this.currentlyRunningStatementFuture = null;
+          }
+        }
+        if (isSingleUse()) {
+          endUnitOfWorkSpan();
+        }
+        statementFuture.setException(t);
+        throw t;
+      }
       final ApiFuture<T> future =
           ApiFutures.catching(
               f,
@@ -417,25 +437,36 @@ public <ReqT, RespT> ApiCallContext configure(
                 throw SpannerExceptionFactory.asSpannerException(input);
               },
               MoreExecutors.directExecutor());
-      synchronized (this) {
-        this.currentlyRunningStatementFuture = future;
-      }
-      future.addListener(
-          new Runnable() {
+      ApiFutures.addCallback(
+          future,
+          new ApiFutureCallback<T>() {
             @Override
-            public void run() {
-              synchronized (this) {
-                if (currentlyRunningStatementFuture == future) {
-                  currentlyRunningStatementFuture = null;
-                }
-              }
-              if (isSingleUse()) {
-                endUnitOfWorkSpan();
+            public void onFailure(Throwable t) {
+              statementFuture.setException(t);
+            }
+
+            @Override
+            public void onSuccess(T result) {
+              statementFuture.set(result);
+            }
+          },
+          MoreExecutors.directExecutor());
+      statementFuture.addListener(
+          () -> {
+            if (statementFuture.isCancelled()) {
+              future.cancel(true);
+            }
+            synchronized (AbstractBaseUnitOfWork.this) {
+              if (currentlyRunningStatementFuture == statementFuture) {
+                currentlyRunningStatementFuture = null;
               }
             }
+            if (isSingleUse()) {
+              endUnitOfWorkSpan();
+            }
           },
           MoreExecutors.directExecutor());
-      return future;
+      return statementFuture;
     }
   }