arimu1 opened a new pull request, #40107: URL: https://github.com/apache/beam/pull/40107
## Summary - When `BigtableIO.readChangeStream()` uses `ExistingPipelineOptions.RESUME_OR_NEW`, a re-executed `InitializeDoFn` (e.g. Dataflow bundle retry) could see Detect New Partition metadata written by the same job and take the resume path, emitting a second `InitialPipelineState` and starting duplicate change-stream consumers. - Assign a stable per-pipeline-run id at graph construction time, persist it in the metadata table on successful initialization, and skip `InitializeDoFn` output when the stored id matches the current run. A genuinely new job still resumes when the stored id differs. Fixes #39970 ## Test plan - [x] Added `InitializeDoFnTest.testInitializeSkipsDuplicatePipelineRun` (emulator) - [ ] `./gradlew :sdks:java:io:google-cloud-platform:test --tests org.apache.beam.sdk.io.gcp.bigtable.changestreams.dofn.InitializeDoFnTest` (local run hit disk pressure on cold clone; CI expected to run) Made with [Cursor](https://cursor.com) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
