[Bug] Incorrect END_INPUT recovery may cause data loss and stale watermark
- Dominant language
- Java
- Stars
- 3.4k
- Forks
- 1.4k
- Avg merge
- 1d 11h
- Merged PRs (30d)
- 396
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.
### Paimon version
1.1.1
### Compute Engine
Flink1.19
### Minimal reproduce step
## Problem
`Long.MAX_VALUE` is used as the END_INPUT commit identifier. Different writer subtasks may reach `endInput()` at different times, so a checkpoint can contain only a subset of the final END_INPUT committables.
Two correctness issues exist around such pending END_INPUT state.
### 1. Partial END_INPUT committable may be committed during recovery and cause data loss
Consider a job with multiple writer subtasks:
```text
writer-0 reaches endInput
-> emits END_INPUT committable(MAX)
checkpoint N
-> committer state contains writer-0's MAX
writer-1 has not reached endInput yet
failover
```
During recovery, the existing implementation immediately calls `filterAndCommit` for restored committables. Therefore, the partial MAX containing only writer-0's data can be committed before writer-1 produces its END_INPUT committable.
Once that partial commit succeeds, the latest Paimon snapshot already has:
```text
commitIdentifier = Long.MAX_VALUE
```
`filterAndCommit` filters by commit identifier rather than comparing the actual payload. Later, when writer-1 reaches `endInput()`, its END_INPUT committable also has identifier `Long.MAX_VALUE`, so it is filtered out and never committed. This results in data loss.
The same completeness invariant must also hold at runtime. Remaining END_INPUT committables may already have reached `CommitterOperator.processElement()` but still be buffered in `inputs`. If `endInput=true` becomes visible before `pollInputs()` materializes them into the MAX bucket, a delayed checkpoint-complete notification may again commit only a partial MAX.
The required invariant is:
```text
Any Long.MAX_VALUE committable that is allowed to commit
must represent the complete logical END_INPUT of all writer subtasks.
```
### 2. Pending END_INPUT watermark may become stale after recovery
When an existing END_INPUT committable is merged with later committables, its watermark is not updated.
For example:
```text
partial MAX(watermark=1024)
-> restore
-> runtime watermark advances to 2048
-> merge remaining MAX
-> checkpoint
-> restore again
```
The restored END_INPUT committable may still carry watermark `1024` instead of `2048`, because the merge path appends the payload but does not persist the newer watermark.
Configured END_INPUT watermark has a similar issue when the MAX committable already exists and there are no new buffered inputs to trigger a merge.
## Proposed direction
Handle the two problems independently:
1. Prevent incomplete END_INPUT committables from being committed during recovery or runtime finalization.
2. Preserve and correctly update END_INPUT watermark across merge, checkpoint, and recovery.
### What doesn't meet your expectations?
none
### Anything else?
_No response_
### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by tracing CommitterOperator.processElement(), pollInputs(), endInput(), and filterAndCommit() through checkpoint recovery and runtime finalization. Verify that Long.MAX_VALUE committables are complete before commit, and that merged or configured END_INPUT state preserves the newest watermark across checkpoint and recovery.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100