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

Inizia tracciando CommitterOperator.processElement(), pollInputs(), endInput() e filterAndCommit() attraverso il recupero dal checkpoint e la finalizzazione del runtime. Verifica che i committables di Long.MAX_VALUE siano completi prima del commit e che lo stato END_INPUT unito o configurato conservi il watermark più recente durante il checkpoint e il recupero.

Search before asking

• I searched in the 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:

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:

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:

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:

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?

• I'm willing to submit a PR!