gschlueter-jaconi opened a new issue, #39970:
URL: https://github.com/apache/beam/issues/39970
### 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
```java
BigtableIO.readChangeStream()
.withProjectId(projectId)
.withInstanceId(instanceId)
.withTableId(tableId)
.withAppProfileId(appProfileId)
.withMetadataTableTableId(metadataTableId)
.withChangeStreamName(changeStreamName)
.withExistingPipelineOptions(
BigtableIO.ExistingPipelineOptions.RESUME_OR_NEW);
```
## Observed behavior
1. The job starts normally and consumes every event once.
2. After approximately 30 minutes, the running job logs:
```text
Resuming from previous pipeline with low watermark of ...
```
3. Every subsequent event is then consumed exactly twice.
4. No second job is started and the worker count remains one.
5. The duplicate consumption is persistent, not just a short overlap during
a checkpoint.
6. 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:
```java
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
1. 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.
2. 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
- [ ] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [x] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
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]
