[Data Validation] Add Spanner failure injection test + Base class

v2/failure-injection-policies/src/main/java/com/google/cloud/teleport/v2/failureinjection/InitialLimitedDurationErrorInjectionPolicy.java

@@ -22,6 +22,7 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.time.format.DateTimeParseException;
+import java.util.concurrent.atomic.AtomicLong;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -37,12 +38,12 @@ public class InitialLimitedDurationErrorInjectionPolicy
       LoggerFactory.getLogger(InitialLimitedDurationErrorInjectionPolicy.class);
   private static final long serialVersionUID = 1L;
 
-  private Instant startTime;
+  private static volatile Instant startTime = null;
+  private static final AtomicLong callCount = new AtomicLong(0);
   private final Duration injectionDuration;
   private final String effectiveDurationParameter;
   private String errorCodeToBeInjected;
   private Clock clock;
-  private long callCount;
 
   private static final String DEFAULT_DURATION = "PT10M";
   private static final String DURATION_FIELD_IN_OBJECT = "duration";
@@ -124,22 +125,20 @@ public InitialLimitedDurationErrorInjectionPolicy(JsonNode inputParameter, Clock
    */
   @Override
   public boolean shouldInjectionError() {
-    if (this.startTime == null) {
-      synchronized (this) {
-        if (this.startTime == null) {
-          this.startTime = Instant.now(clock);
+    if (startTime == null) {
+      synchronized (InitialLimitedDurationErrorInjectionPolicy.class) {
+        if (startTime == null) {
+          startTime = Instant.now(clock);
           LOG.info(
               "First call detected. Errors will be injected for {} starting from {}.",
               this.injectionDuration,
-              this.startTime);
+              startTime);
         }
       }
     }
-    synchronized (this) {
-      ++callCount;
-    }
+    long currentCallCount = callCount.incrementAndGet();
 
-    if (callCount < INITIAL_ALLOWED_CALLS_COUNT) {
+    if (currentCallCount < INITIAL_ALLOWED_CALLS_COUNT) {
       return false;
     }
 
@@ -186,6 +185,11 @@ void setClockForTesting(Clock clock) {
     this.clock = clock;
   }
 
+  public static void resetForTesting() {
+    startTime = null;
+    callCount.set(0);
+  }
+
   @Override
   public String toString() {
     return "InitialLimitedDurationErrorInjectionPolicy{"

v2/failure-injection-policies/src/test/java/com/google/cloud/teleport/v2/failureinjection/InitialLimitedDurationErrorInjectionPolicyTest.java

@@ -28,10 +28,16 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.time.ZoneOffset;
+import org.junit.Before;
 import org.junit.Test;
 
 public class InitialLimitedDurationErrorInjectionPolicyTest {
 
+  @Before
+  public void setUp() {
+    InitialLimitedDurationErrorInjectionPolicy.resetForTesting();
+  }
+
   private ObjectNode createInputObject(String duration) {
     ObjectNode node = JsonNodeFactory.instance.objectNode();
     if (duration != null) {
@@ -268,4 +274,26 @@ public void shouldInjectError_startTimeDoesNotChangeAfterFirstCall() {
     Instant thirdStartTime = policy.getStartTime();
     assertEquals("Start time should still not change", firstStartTime, thirdStartTime);
   }
+
+  @Test
+  public void constructor_shouldParseErrorCode() {
+    ObjectNode input = JsonNodeFactory.instance.objectNode();
+    input.put("duration", "PT5S");
+    input.put("errorCode", "UNAVAILABLE");
+    Clock clock = Clock.fixed(Instant.EPOCH, ZoneOffset.UTC);
+    InitialLimitedDurationErrorInjectionPolicy policy =
+        new InitialLimitedDurationErrorInjectionPolicy(input, clock);
+
+    assertEquals("UNAVAILABLE", policy.getErrorCodeToBeInjected());
+  }
+
+  @Test
+  public void constructor_shouldUseDefaultErrorCodeIfBlank() {
+    ObjectNode input = JsonNodeFactory.instance.objectNode();
+    input.put("duration", "PT5S");
+    input.put("errorCode", "  ");
+    Clock clock = Clock.fixed(Instant.EPOCH, ZoneOffset.UTC);
+    InitialLimitedDurationErrorInjectionPolicy policy =
+        new InitialLimitedDurationErrorInjectionPolicy(input, clock);
+  }
 }

v2/gcs-spanner-dv/pom.xml

@@ -95,9 +95,44 @@
                          parent POM(s). -->
                         <exclude>com/google/cloud/teleport/v2/dto/**</exclude>
                         <exclude>com/google/cloud/teleport/v2/constants/**</exclude>
+                        <exclude>com/google/cloud/teleport/v2/templates/GCSSpannerDV.class</exclude>
                     </excludes>
                 </configuration>
             </plugin>
         </plugins>
     </build>
+    <profiles>
+        <profile>
+            <id>useRealSpanner</id>
+            <activation>
+                <activeByDefault>true</activeByDefault>
+                <property>
+                    <name>!activateFailureInjection</name>
+                </property>
+            </activation>
+            <dependencies>
+                <dependency>
+                    <groupId>com.google.cloud.teleport.v2</groupId>
+                    <artifactId>real-spanner-service</artifactId>
+                    <version>${project.version}</version>
+                </dependency>
+            </dependencies>
+        </profile>
+        <profile>
+            <id>failureInjectionTest</id>
+            <activation>
+                <property>
+                    <name>activateFailureInjection</name>
+                    <value>true</value>
+                </property>
+            </activation>
+            <dependencies>
+                <dependency>
+                    <groupId>com.google.cloud.teleport.v2</groupId>
+                    <artifactId>failure-injected-spanner-service</artifactId>
+                    <version>${project.version}</version>
+                </dependency>
+            </dependencies>
+        </profile>
+    </profiles>
 </project>

v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/templates/GCSSpannerDV.java

@@ -37,6 +37,7 @@
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
+import org.apache.beam.sdk.io.gcp.spanner.SpannerServiceFactoryImpl;
 import org.apache.beam.sdk.options.Default;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
@@ -254,6 +255,16 @@ public interface Options extends PipelineOptions {
     String getTransformationCustomParameters();
 
     void setTransformationCustomParameters(String value);
+
+    @TemplateParameter.Text(
+        order = 16,
+        optional = true,
+        description = "Failure injection parameter",
+        helpText = "Failure injection parameter. Only used for testing.")
+    @Default.String("")
+    String getFailureInjectionParameter();
+
+    void setFailureInjectionParameter(String value);
   }
 
   public static void main(String[] args) {
@@ -327,11 +338,20 @@ public static PipelineResult run(Options options) {
 
   @VisibleForTesting
   static SpannerConfig createSpannerConfig(Options 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 config =
+        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.getFailureInjectionParameter() != null
+        && !options.getFailureInjectionParameter().isEmpty()) {
+      config =
+          SpannerServiceFactoryImpl.createSpannerService(
+              config, options.getFailureInjectionParameter());
+    }
+    return config;
   }
 }

v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVFTBase.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright (C) 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package com.google.cloud.teleport.v2.templates;
+
+import com.google.cloud.teleport.v2.spanner.migrations.transformation.CustomTransformation;
+import com.google.common.io.Resources;
+import java.io.IOException;
+import java.util.Map;
+import org.apache.beam.it.common.PipelineLauncher.LaunchInfo;
+import org.apache.beam.it.common.utils.PipelineUtils;
+import org.apache.beam.it.gcp.dataflow.FlexTemplateDataflowJobResourceManager;
+import org.apache.beam.it.gcp.spanner.SpannerResourceManager;
+
+/**
+ * Base class for gcs-spanner-dv failure injection integration tests.
+ *
+ * <p><strong>Why is this a separate class from {@link GCSSpannerDVITBase}?</strong>
+ *
+ * <p>While {@code GCSSpannerDVITBase} is runner-agnostic and relies on the generic {@code
+ * PipelineLauncher} (allowing tests to run locally via DirectRunner), failure injection testing
+ * explicitly requires building a custom Docker image with the {@code failureInjectionTest} Maven
+ * profile.
+ *
+ * <p>Therefore, tests extending this class are strictly coupled to Dataflow Flex Templates and
+ * bypass the generic launcher in favor of {@link FlexTemplateDataflowJobResourceManager}.
+ */
+public abstract class GCSSpannerDVFTBase extends GCSSpannerDVITBase {
+
+  /**
+   * Launches the Dataflow job with failure injection testing capabilities using
+   * FlexTemplateDataflowJobResourceManager.
+   */
+  protected LaunchInfo launchFTDataflowJob(
+      String testId,
+      String projectId,
+      SpannerResourceManager spannerResourceManager,
+      String bigQueryDataset,
+      String gcsInputDirectory,
+      String sessionFileResourceName,
+      String schemaOverridesFileResourceName,
+      String tableOverrides,
+      String columnOverrides,
+      CustomTransformation customTransformation,
+      String failureInjectionParameter,
+      Map<String, String> jobParameters)
+      throws IOException {
+
+    FlexTemplateDataflowJobResourceManager.Builder flexTemplateBuilder =
+        FlexTemplateDataflowJobResourceManager.builder(testId)
+            .withTemplateName("GCS_Spanner_Data_Validator")
+            .withTemplateModulePath("v2/gcs-spanner-dv")
+            .withAdditionalMavenProfile("failureInjectionTest")
+            .addEnvironmentVariable(
+                "additionalExperiments", java.util.Collections.singletonList("disable_runner_v2"));
+
+    if (failureInjectionParameter != null && !failureInjectionParameter.isEmpty()) {
+      flexTemplateBuilder.addParameter("failureInjectionParameter", failureInjectionParameter);
+    }
+
+    flexTemplateBuilder.addParameter("projectId", projectId);
+    flexTemplateBuilder.addParameter("instanceId", spannerResourceManager.getInstanceId());
+    flexTemplateBuilder.addParameter("databaseId", spannerResourceManager.getDatabaseId());
+    flexTemplateBuilder.addParameter("bigQueryDataset", bigQueryDataset);
+    flexTemplateBuilder.addParameter("gcsInputDirectory", gcsInputDirectory);
+
+    if (sessionFileResourceName != null) {
+      gcsClient.uploadArtifact(
+          "session.json", Resources.getResource(sessionFileResourceName).getPath());
+      flexTemplateBuilder.addParameter("sessionFilePath", getGcsPath("session.json"));
+    }
+
+    if (schemaOverridesFileResourceName != null) {
+      gcsClient.uploadArtifact(
+          "schema_overrides.json",
+          Resources.getResource(schemaOverridesFileResourceName).getPath());
+      flexTemplateBuilder.addParameter(
+          "schemaOverridesFilePath", getGcsPath("schema_overrides.json"));
+    }
+
+    if (tableOverrides != null) {
+      flexTemplateBuilder.addParameter("tableOverrides", tableOverrides);
+    }
+
+    if (columnOverrides != null) {
+      flexTemplateBuilder.addParameter("columnOverrides", columnOverrides);
+    }
+
+    if (customTransformation != null) {
+      flexTemplateBuilder.addParameter(
+          "transformationJarPath", getGcsPath(customTransformation.jarPath()));
+      flexTemplateBuilder.addParameter("transformationClassName", customTransformation.classPath());
+      if (customTransformation.customParameters() != null) {
+        flexTemplateBuilder.addParameter(
+            "transformationCustomParameters", customTransformation.customParameters());
+      }
+    }
+
+    String runId = PipelineUtils.createJobName(testId);
+    flexTemplateBuilder.addParameter("runId", runId);
+    flexTemplateBuilder.addParameter("workerMachineType", "n2-standard-4");
+
+    if (jobParameters != null) {
+      jobParameters.forEach(flexTemplateBuilder::addParameter);
+    }
+
+    return flexTemplateBuilder.build().launchJob();
+  }
+}