[Data Validation] Add Spanner failure injection test + Base class - #4124
[Data Validation] Add Spanner failure injection test + Base class#4124aasthabharill wants to merge 8 commits into
Conversation
aasthabharill
changed the title
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces infrastructure to test the resilience of the GCS-to-Spanner Data Validation pipeline against transient Spanner errors. It adds a dedicated Maven profile for failure injection, updates the pipeline configuration to support error injection parameters, and includes a new integration test that simulates 'UNAVAILABLE' errors to ensure the pipeline's read operations recover correctly. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces failure injection testing capabilities for the gcs-spanner-dv template. It updates the InitialLimitedDurationErrorInjectionPolicy to use static fields for tracking state across instances, adds Maven profiles to switch between real and failure-injected Spanner services, and implements a new integration test (GCSSpannerDVSpannerReadFT) using a Flex Template launcher. Feedback focuses on addressing thread-safety issues caused by synchronizing on this for static fields, preventing state leakage across test runs in the same JVM, and correcting a parameter mismatch in the test base class where testName was used instead of the passed testId.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #4124 +/- ##
============================================
+ Coverage 55.89% 55.93% +0.03%
+ Complexity 7476 7023 -453
============================================
Files 1134 1133 -1
Lines 70135 70098 -37
Branches 8014 8013 -1
============================================
+ Hits 39202 39207 +5
+ Misses 28387 28345 -42
Partials 2546 2546
🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Code Review
This pull request introduces failure injection testing capabilities to the GCS to Spanner Data Validator (gcs-spanner-dv) pipeline. It updates InitialLimitedDurationErrorInjectionPolicy to use static thread-safe fields (volatile and AtomicLong) with class-level synchronization, adds a failureInjectionParameter template option to GCSSpannerDV, and configures Maven profiles to support building with a failure-injected Spanner service. Additionally, it adds integration tests (GCSSpannerDVSpannerReadFT) to verify pipeline resilience against transient Spanner read failures. Feedback on the changes highlights that errorCodeToBeInjected in InitialLimitedDurationErrorInjectionPolicy is not initialized with a default value, which could leave it as null when the input is blank. Furthermore, the newly added test constructor_shouldUseDefaultErrorCodeIfBlank is missing assertions to verify this default behavior.
| private static final AtomicLong callCount = new AtomicLong(0); | ||
| private final Duration injectionDuration; | ||
| private final String effectiveDurationParameter; | ||
| private String errorCodeToBeInjected; |
There was a problem hiding this comment.
The errorCodeToBeInjected field is not initialized with a default value. When the errorCode parameter is blank or missing in the input JSON, the constructor logs that it is using the default DEADLINE_EXCEEDED, but it does not actually assign a value to errorCodeToBeInjected, leaving it as null. Initializing it to Code.DEADLINE_EXCEEDED.name() by default ensures that the policy behaves as documented and avoids potential NullPointerExceptions in the caller.
| private String errorCodeToBeInjected; | |
| private String errorCodeToBeInjected = Code.DEADLINE_EXCEEDED.name(); |
|
|
||
| private Instant startTime; | ||
| private static volatile Instant startTime = null; | ||
| private static final AtomicLong callCount = new AtomicLong(0); |
There was a problem hiding this comment.
Since we made the state static, all the threads on a Dataflow worker are now hitting the exact same counter. So if we were to use synchronized block, every thread would have to acquire the lock and wait. AtomicLong is better as incrementAndGet() allows threads to update the counter without blocking each other.
| synchronized (InitialLimitedDurationErrorInjectionPolicy.class) { | ||
| if (startTime == null) { | ||
| startTime = Instant.now(clock); |
There was a problem hiding this comment.
We are synchronising on the class now instead of the object of the class? Do I understand that right? What was the issue happening earlier?
There was a problem hiding this comment.
What was happening earlier:
The original issue was that Beam actually deserializes multiple instances of this policy per worker (usually one per thread or bundle). When the variables weren't static, each instance got its own isolated timer. This meant the duration kept resetting per-thread instead of applying globally - so i kept getting the same fake exception for each retry ultimately failing the job.
I fixed that by making the state static so it's shared across the worker. But once the state is static, locking on this (the instance) is unsafe because threads using different instances would acquire different locks, leading to race conditions. Synchronizing on the Class object ensures all instances share the exact same lock to initialize the global timer safely.
| List<Mutation> mutations = new ArrayList<>(); | ||
| Instant now = Instant.now().truncatedTo(java.time.temporal.ChronoUnit.MILLIS); | ||
|
|
||
| // We insert 500 rows to ensure Dataflow has enough data to potentially split bundles |
There was a problem hiding this comment.
nit: replace 500 with NUM_RECORDS
aasthabharill
left a comment
aasthabharill
left a comment
There was a problem hiding this comment.
converting to draft as i triggered the failure injection tests using InitialLimitedDurationErrorInjectionPolicy and the bulk and reverse FTs are failing - will debug + add change to trigger FT github workflow on such PRs
2 participants
Successful run on latest code: https://github.com/GoogleCloudPlatform/DataflowTemplates/actions/runs/32717186875/job/97400820802
Dataflow job: https://pantheon.corp.google.com/dataflow/jobs/us-central1/2026-08-24_03_40_35-17616644887397284197;logsSeverity=ERROR;graphView=0?project=span-cloud-teleport-testing&pageState=(%22dfTime%22:(%22l%22:%22dfJobMaxTime%22))&e=PangolinKitchenLaunch::PangolinKitchenEnabled&mods=logs_tg_staging
Errors were correctly simulated but LESS than 4 times so pipeline can recover on retry