What happened?
Environment
- Apache Beam Java SDK 2.75.0
- Google Cloud Dataflow / Portable Runner
- Exactly one active job and one worker
- Stable, explicitly configured
changeStreamName
Configuration
BigtableIO.readChangeStream()
.withProjectId(projectId)
.withInstanceId(instanceId)
.withTableId(tableId)
.withAppProfileId(appProfileId)
.withMetadataTableTableId(metadataTableId)
.withChangeStreamName(changeStreamName)
.withExistingPipelineOptions(
BigtableIO.ExistingPipelineOptions.RESUME_OR_NEW);Observed behavior
-
The job starts normally and consumes every event once.
-
After approximately 30 minutes, the running job logs:
Resuming from previous pipeline with low watermark of ...
-
Every subsequent event is then consumed exactly twice.
-
No second job is started and the worker count remains one.
-
The duplicate consumption is persistent, not just a short overlap during a checkpoint.
-
Omitting withExistingPipelineOptions(...), thereby using the default FAIL_IF_EXISTS, prevents the problem.
Expected behavior
Retries, checkpointing, or repeated initialization within the same active job must not cause it to resume itself or create another consumer for partitions already being consumed.
Suspected cause
Dataflow can re-execute InitializeDoFn. After the current job has created its metadata, RESUME_OR_NEW appears to interpret that metadata as belonging to a previous pipeline. The observed Resuming from previous pipeline... log line indicates that this branch is executed while the original consumer is still active.
ResumeFromPreviousPipelineAction restores the existing partition UUIDs. The locking code in MetadataTableDao.lockAndRecordPartition() considers an existing lock with the same UUID successfully held:
if (doHoldLock(
partitionRecord.getPartition(),
partitionRecord.getUuid())) {
return true;
}The original and resumed readers can therefore both pass the lock check and open change-stream RPCs for the same partitions.
The workaround supports this explanation. With the default FAIL_IF_EXISTS, repeated initialization detects the existing metadata and returns without emitting another InitialPipelineState, preventing another downstream consumer chain from starting.
Workaround
Omit withExistingPipelineOptions(...). This prevents duplicate initialization, but it also prevents a legitimate later job from resuming existing metadata, so it is not a complete solution.
Suggested fixes
-
Make initialization idempotent per job execution. Store a unique execution identifier in the metadata and claim initialization atomically. Re-execution with the same identifier should emit nothing, while a genuinely new job should still be allowed to resume.
-
Add a durable checkpoint or reshuffle boundary between InitializeDoFn and the unbounded DetectNewPartitionsDoFn. This should ensure that retries or backup executions of initialization cannot start multiple downstream consumer chains.
Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
Issue Components
What happened?
Environment
changeStreamNameConfiguration
Observed behavior
The job starts normally and consumes every event once.
After approximately 30 minutes, the running job logs:
Every subsequent event is then consumed exactly twice.
No second job is started and the worker count remains one.
The duplicate consumption is persistent, not just a short overlap during a checkpoint.
Omitting
withExistingPipelineOptions(...), thereby using the defaultFAIL_IF_EXISTS, prevents the problem.Expected behavior
Retries, checkpointing, or repeated initialization within the same active job must not cause it to resume itself or create another consumer for partitions already being consumed.
Suspected cause
Dataflow can re-execute
InitializeDoFn. After the current job has created its metadata,RESUME_OR_NEWappears to interpret that metadata as belonging to a previous pipeline. The observedResuming from previous pipeline...log line indicates that this branch is executed while the original consumer is still active.ResumeFromPreviousPipelineActionrestores the existing partition UUIDs. The locking code inMetadataTableDao.lockAndRecordPartition()considers an existing lock with the same UUID successfully held:The original and resumed readers can therefore both pass the lock check and open change-stream RPCs for the same partitions.
The workaround supports this explanation. With the default
FAIL_IF_EXISTS, repeated initialization detects the existing metadata and returns without emitting anotherInitialPipelineState, preventing another downstream consumer chain from starting.Workaround
Omit
withExistingPipelineOptions(...). This prevents duplicate initialization, but it also prevents a legitimate later job from resuming existing metadata, so it is not a complete solution.Suggested fixes
Make initialization idempotent per job execution. Store a unique execution identifier in the metadata and claim initialization atomically. Re-execution with the same identifier should emit nothing, while a genuinely new job should still be allowed to resume.
Add a durable checkpoint or reshuffle boundary between
InitializeDoFnand the unboundedDetectNewPartitionsDoFn. This should ensure that retries or backup executions of initialization cannot start multiple downstream consumer chains.Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
Issue Components