[BUG] Flink StreamWriteOperatorCoordinator fails with HoodieWriteConflictException on restart during recommitInstant
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
## Describe the bug
When a Flink streaming job restarts with pending inflight instants and OCC (Optimistic Concurrency Control) is enabled, `StreamWriteOperatorCoordinator.restoreEvents()` fails with `HoodieWriteConflictException` during `recommitInstant()`. The conflict is a false positive — it detects overlapping file IDs between the inflight instant being recommitted and the job's own prior completed commits.
## To Reproduce
1. Start a Flink streaming upsert job with `hoodie.write.concurrency.mode=OPTIMISTIC_CONCURRENCY_CONTROL` and metadata table enabled
2. Let the job run for several checkpoints (multiple deltacommits on the timeline)
3. Kill the job while an instant is inflight (not yet committed)
4. Restart the job — `restoreEvents()` attempts to recommit the inflight instant
5. Job fails with `HoodieWriteConflictException`
## Expected behavior
The recommit should succeed when the only completed instants on the timeline are from the same writer's prior checkpoints. Conflict resolution should only flag genuinely concurrent commits from other writers.
## Environment
- Hudi version: 1.0.0+ (also affects 1.0.1, 1.1.0, 1.2)
- Flink version: 1.18+
- Storage: HDFS / Cloud (GCS/CFS)
## Stack trace
```
org.apache.hudi.exception.HoodieWriteConflictException: java.util.ConcurrentModificationException: Cannot resolve conflicts for overlapping writes between first operation = ConcurrentOperation(actionState=INFLIGHT, actionType=deltacommit, instantTime=20260615144320397), second operation = ConcurrentOperation(actionState=COMPLETED, actionType=deltacommit, instantTime=20260612230825400)
at o.a.h.client.transaction.SimpleConcurrentFileWritesConflictResolutionStrategy.resolveConflict(SimpleConcurrentFileWritesConflictResolutionStrategy.java:205)
at o.a.h.client.utils.TransactionUtils.lambda$resolveWriteConflictIfAny$0(TransactionUtils.java:94)
at o.a.h.client.utils.TransactionUtils.resolveWriteConflictIfAny(TransactionUtils.java:88)
at o.a.h.client.BaseHoodieClient.resolveWriteConflict(BaseHoodieClient.java:238)
at o.a.h.client.BaseHoodieWriteClient.preCommit(BaseHoodieWriteClient.java:431)
at o.a.h.client.BaseHoodieWriteClient.commitStats(BaseHoodieWriteClient.java:270)
at o.a.h.client.HoodieFlinkWriteClient.commit(HoodieFlinkWriteClient.java:165)
at o.a.h.client.BaseHoodieWriteClient.commit(BaseHoodieWriteClient.java:224)
at o.a.h.sink.StreamWriteOperatorCoordinator.doCommit(StreamWriteOperatorCoordinator.java:677)
at o.a.h.sink.StreamWriteOperatorCoordinator.commitInstant(StreamWriteOperatorCoordinator.java:648)
at o.a.h.sink.StreamWriteOperatorCoordinator.recommitInstant(StreamWriteOperatorCoordinator.java:565)
at o.a.h.sink.StreamWriteOperatorCoordinator.restoreEvents(StreamWriteOperatorCoordinator.java:455)
at o.a.h.sink.StreamWriteOperatorCoordinator.start(StreamWriteOperatorCoordinator.java:265)
at o.a.flink.runtime.operators.coordination.OperatorCoordinatorHolder.start(OperatorCoordinatorHolder.java:185)
```
**PR:** https://github.com/apache/hudi/pull/19023
Contributor guide
No contributing guide indexed for this repository
Research direction
Start in StreamWriteOperatorCoordinator.java at restoreEvents(), recommitInstant(), and commitInstant(), then trace the conflict check through BaseHoodieWriteClient and SimpleConcurrentFileWritesConflictResolutionStrategy.resolveConflict(). Reproduce the restart scenario with OCC and pending inflight instants; done means the same writer’s prior completed checkpoints no longer cause a false conflict while genuine concurrent conflicts remain detected.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 35/100