apache / apache/paimon

[Bug] Incorrect END_INPUT recovery may cause data loss and stale watermark

Open
#9,236 0 comments 0 reactions 0 assignees View on GitHub
bug
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.