ci: create workflow for Spanner failure injection tests (#3763)

darshan-sj · web-flow · commit b2215d1ce998 · 2026-05-06T16:34:31.000+05:30
* ci: create workflow for Spanner failure injection tests

* Correcting error

* Formatting changes

* Removing duplicate cloudProxy parameter

* Corrected step name
diff --git a/.github/workflows/spanner-failure-injection-tests.yml b/.github/workflows/spanner-failure-injection-tests.yml
@@ -0,0 +1,73 @@
+# Copyright 2025 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
+#
+#     https://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.
+
+name: Spanner Failure Injection tests
+
+on:
+  workflow_dispatch:
+    inputs:
+      test_name:
+        description: 'Fully qualified test name or regex to run (e.g. DataStreamToSpannerCDCFT)'
+        type: string
+        required: true
+
+permissions: write-all
+
+jobs:
+  spanner_failure_injection_tests:
+    name: Spanner Failure Injection Tests
+    timeout-minutes: 180
+    runs-on: [ self-hosted, spanner-it ]
+    steps:
+    - name: Checkout Code
+      uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+      with:
+        fetch-depth: 0
+    - name: Setup Environment
+      id: setup-env
+      uses: ./.github/actions/setup-env
+    - name: Run Spanner failure injection tests
+      run: |
+        ./cicd/run-failure-injection-tests \
+        --modules-to-build="SPANNER" \
+        --it-region="us-central1" \
+        --it-project="span-cloud-teleport-testing" \
+        --it-artifact-bucket="span-cloud-teleport-testing-it-gitactions" \
+        --it-private-connectivity="datastream-connect-2" \
+        --it-cloud-proxy-host="10.128.0.16" \
+        --test="${{ github.event.inputs.test_name }}"
+    - name: Upload Test Report
+      uses: actions/upload-artifact@v7
+      if: always() # always run even if the previous step fails
+      with:
+        name: surefire-test-results
+        path: |
+          **/surefire-reports/TEST-*.xml
+          **/surefire-reports/*.html
+          **/surefire-reports/html/**
+        retention-days: 20
+    - name: Failure Injection Test report on GitHub
+      uses: dorny/test-reporter@v3
+      if: always()
+      with:
+        name: Failure Injection Test report on GitHub
+        path: '**/surefire-reports/TEST-*.xml'
+        reporter: java-junit
+        only-summary: 'true'
+        token: ${{ secrets.GITHUB_TOKEN }}
+        fail-on-error: 'false'
+        list-suites: 'failed'
+        list-tests: 'failed'
+    - name: Cleanup Java Environment
+      uses: ./.github/actions/cleanup-java-env
diff --git a/cicd/cmd/run-failure-injection-tests/main.go b/cicd/cmd/run-failure-injection-tests/main.go
@@ -0,0 +1,85 @@
+/*
+ * 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 main
+
+import (
+	"flag"
+	"log"
+
+	"github.com/GoogleCloudPlatform/DataflowTemplates/cicd/internal/flags"
+	"github.com/GoogleCloudPlatform/DataflowTemplates/cicd/internal/workflows"
+)
+
+func main() {
+	flags.RegisterCommonFlags()
+	flags.RegisterItFlags()
+	flag.Parse()
+
+	// Run mvn install before running integration tests
+	mvnFlags := workflows.NewMavenFlags()
+	err := workflows.MvnCleanInstall().Run(
+		mvnFlags.IncludeDependencies(),
+		mvnFlags.SkipDependencyAnalysis(),
+		mvnFlags.SkipCheckstyle(),
+		mvnFlags.SkipJib(),
+		mvnFlags.SkipTests(),
+		mvnFlags.SkipJacoco(),
+		mvnFlags.SkipShade(),
+		mvnFlags.ThreadCount(1),
+		mvnFlags.InternalMaven())
+	if err != nil {
+		log.Fatalf("%v\n", err)
+	}
+
+	// Run integration tests
+	mvnFlags = workflows.NewMavenFlags()
+	err = workflows.MvnVerify().Run(
+		mvnFlags.DoNotIncludeDependencies(),
+		mvnFlags.SkipDependencyAnalysis(),
+		mvnFlags.SkipCheckstyle(),
+		mvnFlags.SkipJib(),
+		mvnFlags.SkipShade(),
+		mvnFlags.RunFailureInjectionTests(),
+		mvnFlags.ThreadCount(flags.ThreadCount()),
+		mvnFlags.IntegrationTestParallelism(flags.IntegrationTestParallelism()),
+		mvnFlags.StaticBigtableInstance("teleport"),
+		mvnFlags.StaticSpannerInstance("teleport"),
+		mvnFlags.InternalMaven(),
+		flags.Region(),
+		flags.Project(),
+		flags.ArtifactBucket(),
+		flags.StageBucket(),
+		flags.HostIp(),
+		flags.PrivateConnectivity(),
+		flags.SpannerHost(),
+		flags.FailureMode(),
+		flags.RetryFailures(),
+		flags.StaticOracleHost(),
+		flags.StaticOracleSysPassword(),
+		flags.CloudProxyHost(),
+		flags.CloudProxyMySqlPort(),
+		flags.CloudProxyPostgresPort(),
+		flags.CloudProxyPassword(),
+		flags.UnifiedWorkerHarnessContainerImage(),
+		flags.CloudProxyPassword(),
+		mvnFlags.SpecificTest(flags.TestToRun()),
+		mvnFlags.FailIfNoTests(flags.TestToRun() != ""))
+	if err != nil {
+		log.Fatalf("%v\n", err)
+	}
+	log.Println("Build Successful!")
+}
diff --git a/cicd/cmd/run-spanner-staging-it-tests/main.go b/cicd/cmd/run-spanner-staging-it-tests/main.go
@@ -76,8 +76,7 @@ func main() {
 		flags.CloudProxyMySqlPort(),
 		flags.CloudProxyPostgresPort(),
 		flags.CloudProxyPassword(),
-		flags.UnifiedWorkerHarnessContainerImage(),
-		flags.CloudProxyPassword())
+		flags.UnifiedWorkerHarnessContainerImage())
 	if err != nil {
 		log.Fatalf("%v\n", err)
 	}
diff --git a/cicd/internal/workflows/maven-workflows.go b/cicd/internal/workflows/maven-workflows.go
@@ -55,6 +55,7 @@ type MavenFlags interface {
 	RunSpannerStagingIntegrationTests() string
 	RunLoadTests() string
 	RunLoadTestObserver() string
+	RunFailureInjectionTests() string
 	ThreadCount(int) string
 	IntegrationTestParallelism(int) string
 	StaticBigtableInstance(string) string
@@ -138,6 +139,10 @@ func (*mvnFlags) RunLoadTestObserver() string {
 	return "-PtemplatesLoadTestObserve"
 }
 
+func (*mvnFlags) RunFailureInjectionTests() string {
+	return "-PfailureInjectionTest"
+}
+
 // The number of modules Maven is going to build in parallel in a multi-module project.
 func (*mvnFlags) ThreadCount(count int) string {
 	return "-T" + strconv.Itoa(count)
diff --git a/pom.xml b/pom.xml
@@ -904,6 +904,34 @@
         </plugins>
       </build>
     </profile>
+    <profile>
+      <id>failureInjectionTest</id>
+      <activation>
+        <activeByDefault>false</activeByDefault>
+      </activation>
+      <properties>
+        <!-- Skip coverage checks, unit tests are skipped -->
+        <jacoco.skip>true</jacoco.skip>
+        <!-- Some modules may yield no tests -->
+        <failIfNoTests>false</failIfNoTests>
+      </properties>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-surefire-plugin</artifactId>
+            <version>${surefire.version}</version>
+            <configuration combine.self="override">
+              <includes>
+                <include>**/*FT.java</include>
+              </includes>
+              <excludedGroups></excludedGroups>
+              <trimStackTrace>false</trimStackTrace>
+            </configuration>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
     <profile>
       <id>pluginOutputDir</id>
       <activation>
diff --git a/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/failureinjectiontesting/DataStreamToSpannerCDCFT.java b/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/failureinjectiontesting/DataStreamToSpannerCDCFT.java
@@ -54,8 +54,23 @@
 import org.junit.runners.JUnit4;
 
 /**
- * Test for verifying correctness of CDC processing in {@link DataStreamToSpanner} DataStream to
- * Spanner template.
+ * Fault Tolerance (FT) test for verifying the robustness and correctness of Change Data Capture
+ * (CDC) processing in the {@link DataStreamToSpanner} Dataflow template.
+ *
+ * <p>Objective: Verify that the pipeline can successfully process and replicate CDC events from a
+ * MySQL source database to a Spanner target database, even when facing persistent network timeouts
+ * and Spanner transaction aborts during peak load.
+ *
+ * <p>Edge cases covered in this test include:
+ *
+ * <ul>
+ *   <li>Handling simulated Spanner transaction timeouts using the {@code
+ *       TransactionTimeoutInjectionPolicy}.
+ *   <li>Replicating large bursts of CDC events (inserts/updates/deletes) simultaneously while
+ *       undergoing the injected failure condition.
+ *   <li>Ensuring the template successfully retries aborted commits and guarantees consistency in
+ *       CDC event processing without data loss.
+ * </ul>
  */
 @Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class})
 @TemplateIntegrationTest(DataStreamToSpanner.class)
@@ -84,30 +99,38 @@ public class DataStreamToSpannerCDCFT extends DataStreamToSpannerFTBase {
    */
   @Before
   public void setUp() throws IOException, InterruptedException {
-    // create Spanner Resources
+    // 1. Create Target Spanner Databases
+    // The target database stores the replicated data. A separate Spanner database is explicitly
+    // created and passed to the Dataflow job as the shadow database (used for internal state
+    // management). Passing a distinct shadow database triggers the
+    // "Separate Shadow Table" distributed transaction algorithm for processing CDC events.
     spannerResourceManager = createSpannerDatabase(SPANNER_DDL_RESOURCE);
     shadowTableSpannerResourceManager =
         SpannerResourceManager.builder("shadow-" + testName, PROJECT, REGION)
             .maybeUseStaticInstance()
             .build();
     shadowTableSpannerResourceManager.ensureUsableAndCreateResources();
-    // create Source Resources
+
+    // 2. Create Source MySQL Database
+    // Set up the upstream MySQL database and create the Users table.
     sourceDBResourceManager = CloudMySQLResourceManager.builder(testName).build();
     sourceDBResourceManager.createTable(
         USERS_TABLE, new JDBCResourceManager.JDBCSchema(USERS_TABLE_MYSQL_DDL, "id"));
     sourceConnectionProfile =
         createMySQLSourceConnectionProfile(sourceDBResourceManager, List.of(USERS_TABLE));
 
-    // create and upload GCS Resources
+    // 3. Create and upload GCS Resources
+    // Set up the GCS bucket where Datastream will write CDC files.
     gcsResourceManager =
         GcsResourceManager.builder(artifactBucketName, getClass().getSimpleName(), credentials)
             .build();
 
-    // create pubsub manager
+    // 4. Create Pub/Sub Resources
+    // Set up Pub/Sub topics and subscriptions to notify Dataflow of new files in GCS.
+    // Separate topics are created for the main data stream and Dead Letter Queue (DLQ).
     pubsubResourceManager = setUpPubSubResourceManager();
 
     String testRootDir = getClass().getSimpleName();
-    // create subscriptions
     String gcsPrefix =
         String.join("/", new String[] {testRootDir, gcsResourceManager.runId(), testName, "cdc"});
     SubscriptionName subscription =
@@ -124,7 +147,8 @@ public void setUp() throws IOException, InterruptedException {
             gcsResourceManager);
     String artifactBucket = TestProperties.artifactBucket();
 
-    // launch datastream
+    // 5. Launch Datastream Stream
+    // Create and start the Datastream stream to replicate data from MySQL to the GCS bucket.
     DatastreamResourceManager.Builder datastreamResourceManagerBuilder =
         DatastreamResourceManager.builder(testName, PROJECT, REGION)
             .setCredentialsProvider(credentialsProvider);
@@ -140,10 +164,34 @@ public void setUp() throws IOException, InterruptedException {
     int numRows = 100;
     int burstIterations = 10000;
 
-    // generate Load
+    // 6. Generate CDC Load
+    // Initialize the source with 100 base rows, then simulate 10,000 random burst mutations
+    // (inserts, updates, deletes) with a probability of 0.5 to generate CDC stream traffic.
     cdcLoadGenerator = new FuzzyCDCLoadGenerator();
     cdcLoadGenerator.generateLoad(numRows, burstIterations, 0.5, sourceDBResourceManager);
 
+    // 7. Configure Dataflow Failure Injection
+    // This sets the TransactionTimeoutInjectionPolicy, which simulates 260-second long stalls
+    // in Spanner transactions over a 30-minute window, causing standard Spanner RPCs to timeout and
+    // abort.
+    //
+    // NOTE: This test is estimated to take around 45-50 minutes to complete (30-minute injection
+    // window + setup + post-recovery processing buffer).
+    //
+    // Since the Datastream stream is started first and then the CDC load is generated before the
+    // Dataflow job is launched, the pipeline will immediately see all the generated CDC events as a
+    // massive burst upon startup.
+    //
+    // During the 30-minute window, the simulated 260-second delay exceeds standard Spanner Java
+    // Client timeout settings (typically 60 seconds for commit RPCs). This results in
+    // DEADLINE_EXCEEDED errors. This policy specifically tests the "Separate Shadow Table"
+    // transaction algorithm, where the shadow database (tracking stream offsets) and the
+    // main database are distinct. The algorithm utilizes nested transactions to ensure
+    // consistency.
+    //
+    // After the 30-minute window expires, failures are no longer injected. The pipeline then
+    // rapidly
+    // processes the remaining backlog and successfully retries any previously failed events.
     FlexTemplateDataflowJobResourceManager.Builder flexTemplateBuilder =
         FlexTemplateDataflowJobResourceManager.builder(testName)
             .withAdditionalMavenProfile("failureInjectionTest")
@@ -153,7 +201,7 @@ public void setUp() throws IOException, InterruptedException {
             .addEnvironmentVariable("numWorkers", NUM_WORKERS)
             .addEnvironmentVariable("maxWorkers", MAX_WORKERS);
 
-    // launch dataflow template
+    // 8. Launch the Dataflow CDC template
     jobInfo =
         launchFwdDataflowJob(
             spannerResourceManager,
@@ -185,11 +233,14 @@ public void cleanUp() throws IOException {
   public void liveMigrationCrossDbTxnCdcTest()
       throws IOException, InterruptedException, SQLException, ExecutionException {
 
-    // Wait for Forward migration job to be in running state
+    // 1. Wait for the Forward migration job to be in a running state
     assertThatPipeline(jobInfo).isRunning();
 
     long expectedRows = sourceDBResourceManager.getRowCount(USERS_TABLE);
-    // Wait for exact number of rows as source to appear in Spanner
+
+    // 2. Wait for the exact number of rows from the source to successfully appear in Spanner.
+    // This checks that despite the induced transaction timeouts, the pipeline eventually
+    // retries and converges to the correct row count.
     ConditionCheck spannerRowCountConditionCheck =
         SpannerRowsCheck.builder(spannerResourceManager, USERS_TABLE)
             .setMinRows((int) expectedRows)
@@ -202,11 +253,13 @@ public void liveMigrationCrossDbTxnCdcTest()
                 createConfig(jobInfo, Duration.ofHours(1)), spannerRowCountConditionCheck);
     assertThatResult(result).meetsConditions();
 
-    // Usually the dataflow finishes processing the events within 10 minutes. Giving 10 more minutes
-    // buffer for the dataflow job to process the events before asserting the results.
+    // 3. Usually the dataflow finishes processing the events within 10 minutes. Giving 10 more
+    // minutes buffer for the dataflow job to process the events and ensure no late-arriving
+    // events alter the state.
     Thread.sleep(600000);
 
-    // Read data from Spanner and assert that it exactly matches with SourceDb
+    // 4. Read data from both Spanner and MySQL and assert that the exact row states match,
+    // validating the content data integrity and consistency between SourceDb and Spanner.
     cdcLoadGenerator.assertRows(spannerResourceManager, sourceDBResourceManager);
   }
 }
diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/failureinjectiontesting/SpannerToSrcDBMySQLCDCFT.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/failureinjectiontesting/SpannerToSrcDBMySQLCDCFT.java
