fix: Fix transaction tag issue with the blind-write

Author: sakthivelmanii

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

@@ -32,6 +32,7 @@
 import com.google.cloud.spanner.SessionClient.SessionOption;
 import com.google.cloud.spanner.TransactionRunnerImpl.TransactionContextImpl;
 import com.google.cloud.spanner.spi.v1.SpannerRpc;
+import com.google.common.base.Strings;
 import com.google.common.base.Ticker;
 import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.MoreExecutors;
@@ -511,6 +512,11 @@ ApiFuture<Transaction> beginTransactionAsync(
     if (sessionReference.getIsMultiplexed() && mutation != null) {
       requestBuilder.setMutationKey(mutation);
     }
+    if (sessionReference.getIsMultiplexed() && !Strings.isNullOrEmpty(transactionOptions.tag())) {
+      requestBuilder.setRequestOptions(RequestOptions.newBuilder()
+              .setTransactionTag(transactionOptions.tag())
+          .build());
+    }
     final BeginTransactionRequest request = requestBuilder.build();
     final ApiFuture<Transaction> requestFuture;
     XGoogSpannerRequestId reqId = this.getRequestIdCreator().nextRequestId(this.getChannel(), 1);

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

@@ -1898,6 +1898,30 @@ public void testReadWriteExecuteQueryWithTag() {
         .isEqualTo("app=spanner,env=test,action=txn");
   }
 
+  @Test
+  public void testBlindWriteWithTransactionTag() {
+    DatabaseClient client =
+        spanner.getDatabaseClient(DatabaseId.of(TEST_PROJECT, TEST_INSTANCE, TEST_DATABASE));
+    String transactionTag = "app=spanner,env=test,action=txn";
+    TransactionRunner runner =
+        client.readWriteTransaction(Options.tag(transactionTag));
+    runner.run(
+        transaction -> {
+          transaction.buffer(Mutation.newInsertBuilder("abc")
+                  .set("id")
+                  .to(1L)
+              .build());
+          return null;
+        });
+
+    List<BeginTransactionRequest> beginTransactionRequests = mockSpanner.getRequestsOfType(BeginTransactionRequest.class);
+    assertThat(beginTransactionRequests).hasSize(1);
+    assertThat(beginTransactionRequests.get(0).getRequestOptions().getTransactionTag()).isEqualTo(transactionTag);
+    List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
+    assertThat(commitRequests).hasSize(1);
+    assertThat(commitRequests.get(0).getRequestOptions().getTransactionTag()).isEqualTo(transactionTag);
+  }
+
   @Test
   public void testReadWriteExecuteReadWithTag() {
     DatabaseClient client =