fix(spanner): prevent statement cancellation race in AbstractBaseUnitOfWork

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

Fix this by tracking cancellation intent via an `isCancelled` flag guarded by `this`, setting it in `cancel()`, and immediately cancelling the future upon registration if cancellation was already requested.

Author: sakthivelmanii

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

@@ -105,6 +105,9 @@ public String getMessage() {
   @GuardedBy("this")
   private volatile Future<?> currentlyRunningStatementFuture = null;
 
+  @GuardedBy("this")
+  private boolean isCancelled = false;
+
   enum InterceptorsUsage {
     INVOKE_INTERCEPTORS,
     IGNORE_INTERCEPTORS
@@ -277,6 +280,7 @@ StatementTimeout getStatementTimeout() {
   @Override
   public void cancel() {
     synchronized (this) {
+      this.isCancelled = true;
       if (currentlyRunningStatementFuture != null
           && !currentlyRunningStatementFuture.isDone()
           && !currentlyRunningStatementFuture.isCancelled()) {
@@ -419,14 +423,18 @@ public <ReqT, RespT> ApiCallContext configure(
               MoreExecutors.directExecutor());
       synchronized (this) {
         this.currentlyRunningStatementFuture = future;
+        if (this.isCancelled) {
+          future.cancel(true);
+        }
       }
       future.addListener(
           new Runnable() {
             @Override
             public void run() {
-              synchronized (this) {
+              synchronized (AbstractBaseUnitOfWork.this) {
                 if (currentlyRunningStatementFuture == future) {
                   currentlyRunningStatementFuture = null;
+                  isCancelled = false;
                 }
               }
               if (isSingleUse()) {