Exposing spannerio maxCommitDelay as a DF job parameter. (#2928)

set maxCommitDelay=-1 to use the default setting

---------

Co-authored-by: Sandeep Mishra <iamsandeep@google.com>

Author: sm745052

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/SourceDbToSpannerOptions.java

@@ -415,4 +415,18 @@ public interface SourceDbToSpannerOptions extends CommonTemplateOptions {
   String getFailureInjectionParameter();
 
   void setFailureInjectionParameter(String value);
+
+  @TemplateParameter.Text(
+      order = 33,
+      optional = true,
+      description =
+          "Maximum commit delay time (in milliseconds) to optimize write throughput in Spanner. Reference https://cloud.google.com/spanner/docs/throughput-optimized-writes",
+      helpText =
+          "Maximum commit delay time to optimize write throughput in Spanner. Reference https://cloud.google.com/spanner/docs/throughput-optimized-writes."
+              + "Set -1 to let spanner choose the default. Set to a positive value to override for best suited tradeoff of throughput vs latency."
+              + "Defaults to -1.")
+  @Default.Long(-1)
+  Long getMaxCommitDelay();
+
+  void setMaxCommitDelay(Long value);
 }

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/SourceDbToSpanner.java

@@ -131,11 +131,16 @@ private static PipelineResult executeJdbcMigration(
 
   @VisibleForTesting
   static SpannerConfig createSpannerConfig(SourceDbToSpannerOptions options) {
-    return SpannerConfig.create()
-        .withProjectId(ValueProvider.StaticValueProvider.of(options.getProjectId()))
-        .withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()))
-        .withInstanceId(ValueProvider.StaticValueProvider.of(options.getInstanceId()))
-        .withDatabaseId(ValueProvider.StaticValueProvider.of(options.getDatabaseId()))
-        .withRpcPriority(ValueProvider.StaticValueProvider.of(options.getSpannerPriority()));
+    SpannerConfig spannerConfig =
+        SpannerConfig.create()
+            .withProjectId(ValueProvider.StaticValueProvider.of(options.getProjectId()))
+            .withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost()))
+            .withInstanceId(ValueProvider.StaticValueProvider.of(options.getInstanceId()))
+            .withDatabaseId(ValueProvider.StaticValueProvider.of(options.getDatabaseId()))
+            .withRpcPriority(ValueProvider.StaticValueProvider.of(options.getSpannerPriority()));
+    if (options.getMaxCommitDelay() >= 0) {
+      spannerConfig = spannerConfig.withMaxCommitDelay(options.getMaxCommitDelay());
+    }
+    return spannerConfig;
   }
 }

v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/SourceDbToSpannerTest.java

@@ -15,12 +15,14 @@
  */
 package com.google.cloud.teleport.v2.templates;
 
+import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import com.google.cloud.teleport.v2.options.SourceDbToSpannerOptions;
 import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
+import org.joda.time.Duration;
 import org.junit.Test;
 import org.mockito.Mockito;
 
@@ -34,10 +36,17 @@ public void testCreateSpannerConfig() {
     when(mockOptions.getSpannerHost()).thenReturn("testHost");
     when(mockOptions.getInstanceId()).thenReturn("testInstance");
     when(mockOptions.getDatabaseId()).thenReturn("testDatabaseId");
+    when(mockOptions.getMaxCommitDelay()).thenReturn(-1L).thenReturn(42L);
 
+    // For first set of mocks.
     SpannerConfig config = SourceDbToSpanner.createSpannerConfig(mockOptions);
     assertEquals(config.getProjectId().get(), "testProject");
     assertEquals(config.getInstanceId().get(), "testInstance");
     assertEquals(config.getDatabaseId().get(), "testDatabaseId");
+    assertThat(config.getMaxCommitDelay()).isNull();
+
+    // For second set of mocks.
+    config = SourceDbToSpanner.createSpannerConfig(mockOptions);
+    assertEquals(config.getMaxCommitDelay().get(), Duration.millis(42L));
   }
 }