feat: support max_commit_delay in Connection API (#2954)

* feat: support max_commit_delay in Connection API

Adds support for max_commit_delay to the Connection API:
1. Adds a setMaxCommitDelay(Duration) method to Connection
2. Adds a maxCommitDelay connection URL property
3. Adds a SET MAX_COMMIT_DELAY=<duration> SQL statement

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

---------

Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

Author: olavloite

google-cloud-spanner/clirr-ignored-differences.xml

@@ -631,4 +631,16 @@
     <className>com/google/cloud/spanner/connection/Connection</className>
     <method>void setDirectedRead(com.google.spanner.v1.DirectedReadOptions)</method>
   </difference>
+
+  <!-- Added MaxCommitDelay -->
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/spanner/connection/Connection</className>
+    <method>java.time.Duration getMaxCommitDelay()</method>
+  </difference>
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/spanner/connection/Connection</className>
+    <method>void setMaxCommitDelay(java.time.Duration)</method>
+  </difference>
 </differences>

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

@@ -41,6 +41,7 @@
 import com.google.spanner.v1.DirectedReadOptions;
 import com.google.spanner.v1.ExecuteBatchDmlRequest;
 import com.google.spanner.v1.ResultSetStats;
+import java.time.Duration;
 import java.util.Iterator;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
@@ -556,6 +557,16 @@ default String getOptimizerStatisticsPackage() {
   /** @return true if this connection requests commit statistics from Cloud Spanner */
   boolean isReturnCommitStats();
 
+  /** Sets the max_commit_delay that will be applied to commit requests from this connection. */
+  default void setMaxCommitDelay(Duration maxCommitDelay) {
+    throw new UnsupportedOperationException("Unimplemented");
+  }
+
+  /** Returns the max_commit_delay that will be applied to commit requests from this connection. */
+  default Duration getMaxCommitDelay() {
+    throw new UnsupportedOperationException("Unimplemented");
+  }
+
   /**
    * Sets the priority to use for RPCs executed by this connection..
    *

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

@@ -56,6 +56,7 @@
 import com.google.spanner.v1.DirectedReadOptions;
 import com.google.spanner.v1.ExecuteSqlRequest.QueryOptions;
 import com.google.spanner.v1.ResultSetStats;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -245,6 +246,8 @@ static UnitOfWorkType of(TransactionMode transactionMode) {
   private String transactionTag;
   private String statementTag;
 
+  private Duration maxCommitDelay;
+
   /** Create a connection and register it in the SpannerPool. */
   ConnectionImpl(ConnectionOptions options) {
     Preconditions.checkNotNull(options);
@@ -273,6 +276,7 @@ static UnitOfWorkType of(TransactionMode transactionMode) {
     this.autoPartitionMode = options.isAutoPartitionMode();
     this.maxPartitions = options.getMaxPartitions();
     this.maxPartitionedParallelism = options.getMaxPartitionedParallelism();
+    this.maxCommitDelay = options.getMaxCommitDelay();
     this.ddlClient = createDdlClient();
     setDefaultTransactionOptions();
   }
@@ -791,6 +795,18 @@ public boolean isReturnCommitStats() {
     return this.returnCommitStats;
   }
 
+  @Override
+  public void setMaxCommitDelay(Duration maxCommitDelay) {
+    ConnectionPreconditions.checkState(!isClosed(), CLOSED_ERROR_MSG);
+    this.maxCommitDelay = maxCommitDelay;
+  }
+
+  @Override
+  public Duration getMaxCommitDelay() {
+    ConnectionPreconditions.checkState(!isClosed(), CLOSED_ERROR_MSG);
+    return this.maxCommitDelay;
+  }
+
   @Override
   public void setDelayTransactionStartUntilFirstWrite(
       boolean delayTransactionStartUntilFirstWrite) {
@@ -1614,6 +1630,7 @@ UnitOfWork createNewUnitOfWork(boolean isInternalMetadataQuery) {
           .setReadOnlyStaleness(readOnlyStaleness)
           .setAutocommitDmlMode(autocommitDmlMode)
           .setReturnCommitStats(returnCommitStats)
+          .setMaxCommitDelay(maxCommitDelay)
           .setStatementTimeout(statementTimeout)
           .withStatementExecutor(statementExecutor)
           .build();
@@ -1636,6 +1653,7 @@ UnitOfWork createNewUnitOfWork(boolean isInternalMetadataQuery) {
               .setRetryAbortsInternally(retryAbortsInternally)
               .setSavepointSupport(savepointSupport)
               .setReturnCommitStats(returnCommitStats)
+              .setMaxCommitDelay(maxCommitDelay)
               .setTransactionRetryListeners(transactionRetryListeners)
               .setStatementTimeout(statementTimeout)
               .withStatementExecutor(statementExecutor)

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

@@ -44,6 +44,7 @@
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.net.URL;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -357,6 +358,9 @@ private static String generateGuardedConnectionPropertyError(
                   ConnectionProperty.createStringProperty(
                       OPTIMIZER_STATISTICS_PACKAGE_PROPERTY_NAME, ""),
                   ConnectionProperty.createBooleanProperty("returnCommitStats", "", false),
+                  ConnectionProperty.createStringProperty(
+                      "maxCommitDelay",
+                      "The maximum commit delay in milliseconds that should be applied to commit requests from this connection."),
                   ConnectionProperty.createBooleanProperty(
                       "autoConfigEmulator",
                       "Automatically configure the connection to try to connect to the Cloud Spanner emulator (true/false). "
@@ -689,6 +693,7 @@ public static Builder newBuilder() {
   private final String userAgent;
   private final QueryOptions queryOptions;
   private final boolean returnCommitStats;
+  private final Long maxCommitDelay;
   private final boolean autoConfigEmulator;
   private final Dialect dialect;
   private final RpcPriority rpcPriority;
@@ -744,6 +749,7 @@ private ConnectionOptions(Builder builder) {
     queryOptionsBuilder.setOptimizerStatisticsPackage(parseOptimizerStatisticsPackage(this.uri));
     this.queryOptions = queryOptionsBuilder.build();
     this.returnCommitStats = parseReturnCommitStats(this.uri);
+    this.maxCommitDelay = parseMaxCommitDelay(this.uri);
     this.autoConfigEmulator = parseAutoConfigEmulator(this.uri);
     this.dialect = parseDialect(this.uri);
     this.usePlainText = this.autoConfigEmulator || parseUsePlainText(this.uri);
@@ -1074,6 +1080,27 @@ static boolean parseReturnCommitStats(String uri) {
     return Boolean.parseBoolean(value);
   }
 
+  @VisibleForTesting
+  static Long parseMaxCommitDelay(String uri) {
+    String value = parseUriProperty(uri, "maxCommitDelay");
+    try {
+      Long millis = value == null ? null : Long.valueOf(value);
+      if (millis != null && millis < 0L) {
+        throw SpannerExceptionFactory.newSpannerException(
+            ErrorCode.INVALID_ARGUMENT, "maxCommitDelay must be >=0");
+      }
+      return millis;
+    } catch (NumberFormatException numberFormatException) {
+      throw SpannerExceptionFactory.newSpannerException(
+          ErrorCode.INVALID_ARGUMENT,
+          "Invalid value for maxCommitDelay: "
+              + value
+              + "\n"
+              + "The value must be a positive integer indicating the number of "
+              + "milliseconds to use as the max delay.");
+    }
+  }
+
   static boolean parseAutoConfigEmulator(String uri) {
     String value = parseUriProperty(uri, "autoConfigEmulator");
     return Boolean.parseBoolean(value);
@@ -1405,6 +1432,11 @@ public boolean isReturnCommitStats() {
     return returnCommitStats;
   }
 
+  /** The max_commit_delay that should be applied to commit operations on this connection. */
+  public Duration getMaxCommitDelay() {
+    return maxCommitDelay == null ? null : Duration.ofMillis(maxCommitDelay);
+  }
+
   /**
    * Whether connections created by this {@link ConnectionOptions} will automatically try to connect
    * to the emulator using the default host/port of the emulator, and automatically create the

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

@@ -82,6 +82,10 @@ interface ConnectionStatementExecutor {
 
   StatementResult statementShowReturnCommitStats();
 
+  StatementResult statementSetMaxCommitDelay(Duration maxCommitDelay);
+
+  StatementResult statementShowMaxCommitDelay();
+
   StatementResult statementSetDelayTransactionStartUntilFirstWrite(
       Boolean delayTransactionStartUntilFirstWrite);
 

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

@@ -29,6 +29,7 @@
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_DEFAULT_TRANSACTION_ISOLATION;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_DELAY_TRANSACTION_START_UNTIL_FIRST_WRITE;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_DIRECTED_READ;
+import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_MAX_COMMIT_DELAY;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_MAX_PARTITIONED_PARALLELISM;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_MAX_PARTITIONS;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SET_OPTIMIZER_STATISTICS_PACKAGE;
@@ -51,6 +52,7 @@
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_DATA_BOOST_ENABLED;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_DELAY_TRANSACTION_START_UNTIL_FIRST_WRITE;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_DIRECTED_READ;
+import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_MAX_COMMIT_DELAY;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_MAX_PARTITIONED_PARALLELISM;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_MAX_PARTITIONS;
 import static com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType.SHOW_OPTIMIZER_STATISTICS_PACKAGE;
@@ -343,6 +345,26 @@ public StatementResult statementShowReturnCommitStats() {
         SHOW_RETURN_COMMIT_STATS);
   }
 
+  @Override
+  public StatementResult statementSetMaxCommitDelay(Duration duration) {
+    getConnection()
+        .setMaxCommitDelay(
+            duration == null || duration.equals(Duration.getDefaultInstance())
+                ? null
+                : java.time.Duration.ofSeconds(duration.getSeconds(), duration.getNanos()));
+    return noResult(SET_MAX_COMMIT_DELAY);
+  }
+
+  @Override
+  public StatementResult statementShowMaxCommitDelay() {
+    return resultSet(
+        "MAX_COMMIT_DELAY",
+        getConnection().getMaxCommitDelay() == null
+            ? null
+            : getConnection().getMaxCommitDelay().toMillis() + "ms",
+        SHOW_MAX_COMMIT_DELAY);
+  }
+
   @Override
   public StatementResult statementSetDelayTransactionStartUntilFirstWrite(
       Boolean delayTransactionStartUntilFirstWrite) {

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

@@ -57,6 +57,7 @@
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.spanner.v1.SpannerGrpc;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
@@ -111,6 +112,7 @@ static class Builder extends AbstractMultiUseTransaction.Builder<Builder, ReadWr
     private Boolean retryAbortsInternally;
     private boolean delayTransactionStartUntilFirstWrite;
     private boolean returnCommitStats;
+    private Duration maxCommitDelay;
     private SavepointSupport savepointSupport;
     private List<TransactionRetryListener> transactionRetryListeners;
 
@@ -137,6 +139,11 @@ Builder setReturnCommitStats(boolean returnCommitStats) {
       return this;
     }
 
+    Builder setMaxCommitDelay(Duration maxCommitDelay) {
+      this.maxCommitDelay = maxCommitDelay;
+      return this;
+    }
+
     Builder setSavepointSupport(SavepointSupport savepointSupport) {
       this.savepointSupport = savepointSupport;
       return this;
@@ -180,6 +187,9 @@ private TransactionOption[] extractOptions(Builder builder) {
     if (builder.returnCommitStats) {
       numOptions++;
     }
+    if (builder.maxCommitDelay != null) {
+      numOptions++;
+    }
     if (this.transactionTag != null) {
       numOptions++;
     }
@@ -191,6 +201,9 @@ private TransactionOption[] extractOptions(Builder builder) {
     if (builder.returnCommitStats) {
       options[index++] = Options.commitStats();
     }
+    if (builder.maxCommitDelay != null) {
+      options[index++] = Options.maxCommitDelay(builder.maxCommitDelay);
+    }
     if (this.transactionTag != null) {
       options[index++] = Options.tag(this.transactionTag);
     }

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

@@ -46,12 +46,14 @@
 import com.google.cloud.spanner.Type;
 import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
 import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
+import com.google.cloud.spanner.connection.ReadWriteTransaction.Builder;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.spanner.admin.database.v1.DatabaseAdminGrpc;
 import com.google.spanner.v1.SpannerGrpc;
+import java.time.Duration;
 import java.util.concurrent.Callable;
 
 /**
@@ -78,6 +80,7 @@ class SingleUseTransaction extends AbstractBaseUnitOfWork {
   private final TimestampBound readOnlyStaleness;
   private final AutocommitDmlMode autocommitDmlMode;
   private final boolean returnCommitStats;
+  private final Duration maxCommitDelay;
   private final boolean internalMetdataQuery;
   private volatile SettableApiFuture<Timestamp> readTimestamp = null;
   private volatile TransactionRunner writeTransaction;
@@ -92,6 +95,7 @@ static class Builder extends AbstractBaseUnitOfWork.Builder<Builder, SingleUseTr
     private TimestampBound readOnlyStaleness;
     private AutocommitDmlMode autocommitDmlMode;
     private boolean returnCommitStats;
+    private Duration maxCommitDelay;
     private boolean internalMetadataQuery;
 
     private Builder() {}
@@ -135,6 +139,11 @@ Builder setReturnCommitStats(boolean returnCommitStats) {
       return this;
     }
 
+    Builder setMaxCommitDelay(Duration maxCommitDelay) {
+      this.maxCommitDelay = maxCommitDelay;
+      return this;
+    }
+
     Builder setInternalMetadataQuery(boolean internalMetadataQuery) {
       this.internalMetadataQuery = internalMetadataQuery;
       return this;
@@ -164,6 +173,7 @@ private SingleUseTransaction(Builder builder) {
     this.readOnlyStaleness = builder.readOnlyStaleness;
     this.autocommitDmlMode = builder.autocommitDmlMode;
     this.returnCommitStats = builder.returnCommitStats;
+    this.maxCommitDelay = builder.maxCommitDelay;
     this.internalMetdataQuery = builder.internalMetadataQuery;
   }
 
@@ -467,6 +477,9 @@ private TransactionRunner createWriteTransaction() {
     if (returnCommitStats) {
       numOptions++;
     }
+    if (maxCommitDelay != null) {
+      numOptions++;
+    }
     if (numOptions == 0) {
       return dbClient.readWriteTransaction();
     }
@@ -478,6 +491,9 @@ private TransactionRunner createWriteTransaction() {
     if (returnCommitStats) {
       options[index++] = Options.commitStats();
     }
+    if (maxCommitDelay != null) {
+      options[index++] = Options.maxCommitDelay(maxCommitDelay);
+    }
     return dbClient.readWriteTransaction(options);
   }
 

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

@@ -71,6 +71,8 @@ enum ClientSideStatementType {
     SET_OPTIMIZER_STATISTICS_PACKAGE,
     SHOW_RETURN_COMMIT_STATS,
     SET_RETURN_COMMIT_STATS,
+    SHOW_MAX_COMMIT_DELAY,
+    SET_MAX_COMMIT_DELAY,
     SHOW_DELAY_TRANSACTION_START_UNTIL_FIRST_WRITE,
     SET_DELAY_TRANSACTION_START_UNTIL_FIRST_WRITE,
     SHOW_STATEMENT_TAG,

google-cloud-spanner/src/main/resources/com/google/cloud/spanner/connection/ClientSideStatements.json

@@ -112,6 +112,15 @@
 	  "method": "statementShowReturnCommitStats",
 	  "exampleStatements": ["show variable return_commit_stats"]
 	},
+	{
+		"name": "SHOW VARIABLE MAX_COMMIT_DELAY",
+		"executorName": "ClientSideStatementNoParamExecutor",
+		"resultType": "RESULT_SET",
+		"statementType": "SHOW_MAX_COMMIT_DELAY",
+		"regex": "(?is)\\A\\s*show\\s+variable\\s+max_commit_delay\\s*\\z",
+		"method": "statementShowMaxCommitDelay",
+		"exampleStatements": ["show variable max_commit_delay"]
+	},
 	{
 	  "name": "SHOW VARIABLE COMMIT_RESPONSE",
 	  "executorName": "ClientSideStatementNoParamExecutor",
@@ -442,6 +451,21 @@
 	    "converterName": "ClientSideStatementValueConverters$BooleanConverter"
 	  }
 	},
+	{
+		"name": "SET MAX_COMMIT_DELAY = '<duration>'|NULL",
+		"executorName": "ClientSideStatementSetExecutor",
+		"resultType": "NO_RESULT",
+		"statementType": "SET_MAX_COMMIT_DELAY",
+		"regex": "(?is)\\A\\s*set\\s+max_commit_delay\\s*(?:=)\\s*(.*)\\z",
+		"method": "statementSetMaxCommitDelay",
+		"exampleStatements": ["set max_commit_delay=null", "set max_commit_delay='1s'", "set max_commit_delay='100ms'", "set max_commit_delay='10000us'", "set max_commit_delay='9223372036854775807ns'"],
+		"setStatement": {
+			"propertyName": "MAX_COMMIT_DELAY",
+			"separator": "=",
+			"allowedValues": "('(\\d{1,19})(s|ms|us|ns)'|NULL)",
+			"converterName": "ClientSideStatementValueConverters$DurationConverter"
+		}
+	},
 	{
 	  "name": "SET STATEMENT_TAG = '<tag>'",
 	  "executorName": "ClientSideStatementSetExecutor",