Merge branch 'main' into gfe_latency

googleapis · Jan 31, 2025 · 37d2089 · 37d2089
2 parents d336d39 + f4560e5
commit 37d2089
Show file tree

Hide file tree

Showing 20 changed files with 611 additions and 24 deletions.
diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
@@ -5,8 +5,4 @@
 # https://help.github.com/en/github/creating-cloning-and-archiving-repositories/about-code-owners#codeowners-syntax
 
 # The @googleapis/api-spanner-java is the default owner for changes in this repo
-*                       @googleapis/yoshi-java @googleapis/api-spanner-java
-**/*.java               @googleapis/api-spanner-java
-
-# The java-samples-reviewers team is the default owner for samples changes
-samples/**/*.java       @googleapis/java-samples-reviewers @googleapis/api-spanner-java
+*                       @googleapis/yoshi-java @googleapis/spanner-client-libraries-java
diff --git a/.kokoro/continuous/integration-cloud-devel-directpath-enabled.cfg b/.kokoro/continuous/integration-cloud-devel-directpath-enabled.cfg
@@ -27,6 +27,6 @@ env_vars: {
 }
 
 env_vars: {
-    key: "GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS"
+    key: "GOOGLE_SPANNER_ENABLE_DIRECT_ACCESS"
     value: "true"
 }
diff --git a/.kokoro/continuous/integration-cloud-staging-directpath-enabled.cfg b/.kokoro/continuous/integration-cloud-staging-directpath-enabled.cfg
@@ -27,6 +27,6 @@ env_vars: {
 }
 
 env_vars: {
-    key: "GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS"
+    key: "GOOGLE_SPANNER_ENABLE_DIRECT_ACCESS"
     value: "true"
 }
diff --git a/.kokoro/nightly/integration-cloud-devel-directpath-enabled.cfg b/.kokoro/nightly/integration-cloud-devel-directpath-enabled.cfg
@@ -22,7 +22,6 @@ env_vars: {
 }
 
 env_vars: {
-    key: "GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS"
+    key: "GOOGLE_SPANNER_ENABLE_DIRECT_ACCESS"
     value: "true"
 }
-
diff --git a/.kokoro/nightly/integration-cloud-staging-directpath-enabled.cfg b/.kokoro/nightly/integration-cloud-staging-directpath-enabled.cfg
@@ -22,6 +22,6 @@ env_vars: {
 }
 
 env_vars: {
-    key: "GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS"
+    key: "GOOGLE_SPANNER_ENABLE_DIRECT_ACCESS"
     value: "true"
 }
diff --git a/.kokoro/nightly/integration-directpath-enabled.cfg b/.kokoro/nightly/integration-directpath-enabled.cfg
@@ -37,6 +37,6 @@ env_vars: {
 }
 
 env_vars: {
-    key: "GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS"
+    key: "GOOGLE_SPANNER_ENABLE_DIRECT_ACCESS"
     value: "true"
 }
diff --git a/.kokoro/presubmit/integration-directpath-enabled.cfg b/.kokoro/presubmit/integration-directpath-enabled.cfg
@@ -33,6 +33,6 @@ env_vars: {
 }
 
 env_vars: {
-    key: "GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS"
+    key: "GOOGLE_SPANNER_ENABLE_DIRECT_ACCESS"
     value: "true"
-}
+}
diff --git a/README.md b/README.md
@@ -517,6 +517,7 @@ Samples are in the [`samples/`](https://github.com/googleapis/java-spanner/tree/
 | Create Instance Config Sample | [source code](https://github.com/googleapis/java-spanner/blob/main/samples/snippets/src/main/java/com/example/spanner/CreateInstanceConfigSample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-spanner&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/spanner/CreateInstanceConfigSample.java) |
 | Create Instance Example | [source code](https://github.com/googleapis/java-spanner/blob/main/samples/snippets/src/main/java/com/example/spanner/CreateInstanceExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-spanner&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/spanner/CreateInstanceExample.java) |
 | Create Instance Partition Sample | [source code](https://github.com/googleapis/java-spanner/blob/main/samples/snippets/src/main/java/com/example/spanner/CreateInstancePartitionSample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-spanner&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/spanner/CreateInstancePartitionSample.java) |
+| Create Instance With Asymmetric Autoscaling Config Example | [source code](https://github.com/googleapis/java-spanner/blob/main/samples/snippets/src/main/java/com/example/spanner/CreateInstanceWithAsymmetricAutoscalingConfigExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-spanner&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/spanner/CreateInstanceWithAsymmetricAutoscalingConfigExample.java) |
 | Create Instance With Autoscaling Config Example | [source code](https://github.com/googleapis/java-spanner/blob/main/samples/snippets/src/main/java/com/example/spanner/CreateInstanceWithAutoscalingConfigExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-spanner&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/spanner/CreateInstanceWithAutoscalingConfigExample.java) |
 | Create Instance With Processing Units Example | [source code](https://github.com/googleapis/java-spanner/blob/main/samples/snippets/src/main/java/com/example/spanner/CreateInstanceWithProcessingUnitsExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-spanner&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/spanner/CreateInstanceWithProcessingUnitsExample.java) |
 | Create Instance Without Default Backup Schedules Example | [source code](https://github.com/googleapis/java-spanner/blob/main/samples/snippets/src/main/java/com/example/spanner/CreateInstanceWithoutDefaultBackupSchedulesExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-spanner&page=editor&open_in_editor=samples/snippets/src/main/java/com/example/spanner/CreateInstanceWithoutDefaultBackupSchedulesExample.java) |

diff --git a/...nner/src/main/java/com/google/cloud/spanner/AbstractMultiplexedSessionDatabaseClient.java b/...nner/src/main/java/com/google/cloud/spanner/AbstractMultiplexedSessionDatabaseClient.java
@@ -16,10 +16,7 @@
 
 package com.google.cloud.spanner;
 
-import com.google.api.gax.rpc.ServerStream;
 import com.google.cloud.Timestamp;
-import com.google.cloud.spanner.Options.TransactionOption;
-import com.google.spanner.v1.BatchWriteResponse;
 
 /**
  * Base class for the Multiplexed Session {@link DatabaseClient} implementation. Throws {@link
@@ -43,11 +40,4 @@ public String getDatabaseRole() {
   public Timestamp writeAtLeastOnce(Iterable<Mutation> mutations) throws SpannerException {
     return writeAtLeastOnceWithOptions(mutations).getCommitTimestamp();
   }
-
-  @Override
-  public ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
-      Iterable<MutationGroup> mutationGroups, TransactionOption... options)
-      throws SpannerException {
-    throw new UnsupportedOperationException();
-  }
 }
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClientImpl.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClientImpl.java
@@ -189,6 +189,9 @@ public ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
       throws SpannerException {
     ISpan span = tracer.spanBuilder(READ_WRITE_TRANSACTION, commonAttributes, options);
     try (IScope s = tracer.withSpan(span)) {
+      if (canUseMultiplexedSessionsForRW() && getMultiplexedSessionDatabaseClient() != null) {
+        return getMultiplexedSessionDatabaseClient().batchWriteAtLeastOnce(mutationGroups, options);
+      }
       return runWithSessionRetry(session -> session.batchWriteAtLeastOnce(mutationGroups, options));
     } catch (RuntimeException e) {
       span.setStatus(e);

diff --git a/...-spanner/src/main/java/com/google/cloud/spanner/DelayedMultiplexedSessionTransaction.java b/...-spanner/src/main/java/com/google/cloud/spanner/DelayedMultiplexedSessionTransaction.java
@@ -20,12 +20,14 @@
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
+import com.google.api.gax.rpc.ServerStream;
 import com.google.cloud.Timestamp;
 import com.google.cloud.spanner.DelayedReadContext.DelayedReadOnlyTransaction;
 import com.google.cloud.spanner.MultiplexedSessionDatabaseClient.MultiplexedSessionTransaction;
 import com.google.cloud.spanner.Options.TransactionOption;
 import com.google.cloud.spanner.Options.UpdateOption;
 import com.google.common.util.concurrent.MoreExecutors;
+import com.google.spanner.v1.BatchWriteResponse;
 import java.util.concurrent.ExecutionException;
 
 /**
@@ -164,6 +166,22 @@ public CommitResponse writeWithOptions(Iterable<Mutation> mutations, Transaction
     }
   }
 
+  /**
+   * This is a blocking method, as the interface that it implements is also defined as a blocking
+   * method.
+   */
+  @Override
+  public ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
+      Iterable<MutationGroup> mutationGroups, TransactionOption... options)
+      throws SpannerException {
+    SessionReference sessionReference = getSessionReference();
+    try (MultiplexedSessionTransaction transaction =
+        new MultiplexedSessionTransaction(
+            client, span, sessionReference, NO_CHANNEL_HINT, /* singleUse = */ true)) {
+      return transaction.batchWriteAtLeastOnce(mutationGroups, options);
+    }
+  }
+
   @Override
   public TransactionRunner readWriteTransaction(TransactionOption... options) {
     return new DelayedTransactionRunner(

diff --git a/...loud-spanner/src/main/java/com/google/cloud/spanner/MultiplexedSessionDatabaseClient.java b/...loud-spanner/src/main/java/com/google/cloud/spanner/MultiplexedSessionDatabaseClient.java
@@ -22,6 +22,7 @@
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
 import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.rpc.ServerStream;
 import com.google.cloud.Timestamp;
 import com.google.cloud.spanner.Options.TransactionOption;
 import com.google.cloud.spanner.Options.UpdateOption;
@@ -30,6 +31,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.MoreExecutors;
+import com.google.spanner.v1.BatchWriteResponse;
 import com.google.spanner.v1.BeginTransactionRequest;
 import com.google.spanner.v1.RequestOptions;
 import com.google.spanner.v1.Transaction;
@@ -505,6 +507,14 @@ public CommitResponse writeAtLeastOnceWithOptions(
         .writeAtLeastOnceWithOptions(mutations, options);
   }
 
+  @Override
+  public ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
+      Iterable<MutationGroup> mutationGroups, TransactionOption... options)
+      throws SpannerException {
+    return createMultiplexedSessionTransaction(/* singleUse = */ true)
+        .batchWriteAtLeastOnce(mutationGroups, options);
+  }
+
   @Override
   public ReadContext singleUse() {
     return createMultiplexedSessionTransaction(/* singleUse = */ true).singleUse();

diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionImpl.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionImpl.java
@@ -321,6 +321,7 @@ public ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
       throw SpannerExceptionFactory.newSpannerException(e);
     } finally {
       span.end();
+      onTransactionDone();
     }
   }
 

diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/TransactionRunnerImpl.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/TransactionRunnerImpl.java
@@ -326,6 +326,19 @@ private void createTxnAsync(
               }
               res.set(null);
             } catch (ExecutionException e) {
+              SpannerException spannerException = SpannerExceptionFactory.asSpannerException(e);
+              if (spannerException.getErrorCode() == ErrorCode.ABORTED
+                  && session.getIsMultiplexed()
+                  && mutation != null) {
+                // Begin transaction can return ABORTED errors. This can only happen if it included
+                // a mutation key, which again means that this is a mutation-only transaction on a
+                // multiplexed session.
+                span.addAnnotation(
+                    "Transaction Creation Failed with ABORT. Retrying",
+                    e.getCause() == null ? e : e.getCause());
+                createTxnAsync(res, mutation);
+                return;
+              }
               span.addAnnotation(
                   "Transaction Creation Failed", e.getCause() == null ? e : e.getCause());
               res.setException(e.getCause() == null ? e : e.getCause());

diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/MockSpannerServiceImpl.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/MockSpannerServiceImpl.java
@@ -1919,7 +1919,8 @@ private Transaction beginTransaction(
     }
     if (session.getMultiplexed()
         && options.getModeCase() == ModeCase.READ_WRITE
-        && mutationKey != null) {
+        && mutationKey != null
+        && mutationKey != com.google.spanner.v1.Mutation.getDefaultInstance()) {
       // Mutation only case in a read-write transaction.
       builder.setPrecommitToken(getTransactionPrecommitToken(transactionId));
     }
@@ -2023,6 +2024,14 @@ public void commit(CommitRequest request, StreamObserver<CommitResponse> respons
       return;
     }
     sessionLastUsed.put(session.getName(), Instant.now());
+    if (session.getMultiplexed()
+        && !request.hasPrecommitToken()
+        && !request.hasSingleUseTransaction()) {
+      throw Status.INVALID_ARGUMENT
+          .withDescription(
+              "A Commit request for a read-write transaction on a multiplexed session must specify a precommit token.")
+          .asRuntimeException();
+    }
     try {
       commitExecutionTime.simulateExecutionTime(exceptions, stickyGlobalExceptions, freezeLock);
       // Find or start a transaction