[flink] Support end input in coordinator commit

[fishfishfishfishaa]

Purpose

This PR implements the end-of-input handling that was intentionally left out of the core coordinator-commit PR for [PIP-30 #8220 ].

It extends sink.coordinator-commit.enabled to correctly finish and commit bounded inputs. In particular, coordinator commit now supports batch jobs without checkpointing, and correctly handles bounded streaming jobs after all writers reach endInput.

This PR is built on top of the core coordinator-commit implementation. It does not change the default committer-operator path.

What changes

Commit after all writers finish

Each coordinated writer emits a final committable entry with Long.MAX_VALUE as a dedicated end-of-input checkpoint ID when Flink invokes endInput.

The JobManager-side CommittingWriteOperatorCoordinator tracks this entry separately from ordinary checkpoint committables:

• An end-input entry covers all later ordinary checkpoints for that writer, because a finished writer cannot produce more data.
• Once every writer has reported end input, the coordinator performs one final commit through filterAndCommit(..., false, true).
• After the final commit succeeds, repeated end-input events and later checkpoint-complete notifications are ignored.

For streaming jobs with checkpointing enabled, the final commit remains aligned with checkpoint completion. This preserves the normal checkpoint-driven commit model while allowing some subtasks to finish before others.

For batch jobs, where checkpointing is normally disabled, the coordinator commits immediately after receiving end-input entries from all writer subtasks.

Support batch coordinator commit

Coordinator commit previously required streaming mode with checkpointing enabled. This is now relaxed as follows:

• Streaming mode still requires checkpointing.
• Batch mode is supported without checkpointing and commits when all writers reach end input.

The existing coordinator-commit restrictions remain unchanged: it only supports write-only unaware-bucket append tables, and still rejects configurations such as primary-key tables, precommit compaction, auto-tag-for-savepoint, and concurrent checkpoints.

The option documentation is updated accordingly.

Preserve end-input state across recovery

End input is not an ordinary checkpoint: it is newer than every checkpoint and may need to survive restore even if Flink does not invoke endInput again after recovery.

To make this safe:

• Writers persist the final end-input committable in their independent operator state.
• On restore, writers replay the persisted final entry to the coordinator.
• If endInput is invoked again after restore, the writer merges newly produced final committables with the persisted entry and sends one authoritative final entry.
• WriterCommittables treats end input separately from the maximum ordinary checkpoint, so it does not incorrectly constrain ordinary checkpoint alignment.
• Replayed end-input entries are replaceable rather than treated as duplicate ordinary checkpoint reports.

This also handles region failover while the coordinator is still waiting for the remaining writers to finish.

End-input watermark handling

The writer now uses end-input.watermark, when configured, for the final end-input committable.

For ordinary checkpoints after a subtask has finished, that subtask contributes Long.MAX_VALUE to the watermark minimum. This makes a finished writer neutral and prevents it from incorrectly holding back the watermark of active writers.

The final commit uses the configured end-input watermark as expected.

How it works

writer subtask reaches endInput
        |
        v
emit final CheckpointCommittables(Long.MAX_VALUE)
        |
        v
persist final entry in writer operator state
        |
        v
send CommittableEvent to JobManager coordinator
        |
        v
all writer subtasks have reported end input?
        |
        +-- no  --> keep waiting; ended subtasks cover later normal checkpoints
        |
        +-- yes --> perform one final filterAndCommit(..., false, true)

Tests

This PR adds coverage for:

• Batch coordinator commit without checkpointing.
• Streaming coordinator commit with bounded input.
• End-to-end final data commit after input completion.
• End-to-end propagation of end-input.watermark.
• Writer-side emission, persistence, restore, and re-emission of final committables.
• Coordinator-side final-commit behavior when checkpointing is disabled.
• Recovery and region-failover handling for end-input entries.
• WriterCommittables semantics:
  • end input covers later ordinary checkpoints;
  • finished subtasks do not constrain later watermark aggregation;
  • restored state may contain an end-input entry newer than the restored checkpoint;
  • repeated end-input reports replace the authoritative final entry;
  • ordinary cleanup retains end-input state until the final commit.

No production behavior changes when sink.coordinator-commit.enabled is disabled.

[JingsongLi]

• The CoordinatorCommitITCase added in this PR fails in both JDK 8 and JDK 11.
• The coordinator cannot be instantiated in batch mode: Context was not yet initialized.
• In streaming bounded-input mode, the final snapshot watermark remains Long.MIN_VALUE; the configured end-input.watermark=12345 is not committed.
• Additionally, the batch coordinator path bypasses BatchWriteGeneratorTagOperator, and tag.automatic-creation=batch silently fails to create a tag.
• There is still one remaining comment in Chinese in WriterCommittables.

[fishfishfishfishaa]

Regarding the ITCase failure in batch mode: due to the asynchronous architecture of the coordinator, the batch job finishes first, and close() interrupts the final commit. I've adjusted it to use synchronous blocking for batch EndInput events. Since this is not waiting for the last EndInput event — it only waits for in-memory operations — the blocking time is minimal. Moreover, EndInput itself is in the "finalization" phase. Compared to other approaches, I believe this design is simpler and more consistent with the original logic. (Other alternatives include: 1. Modifying close() to wait for all pending commits; 2. Explicitly waiting only for the last one; 3. Adding an ack mechanism.) Additionally, the coordinator Context initialization issue has also been fixed.
In streaming mode, the last snapshot phase was too strict, so end-input.watermark=12345 has been removed. In this scenario, the original CommitterOperator does not promise to create an empty snapshot for writing the watermark either — instead, it relies on committer.forceCreatingSnapshot().

[fishfishfishfishaa]

de, the last snapshot phase was too strict, so end-input.watermark=12345 has been removed. In this scenario, the original CommitterOperator does not promise to create an empty snapshot for writing the watermark either — instead, it relies on committer.forceCreatingSnapshot().

"the batch coordinator path bypasses BatchWriteGeneratorTagOperator, and tag.automatic-creation=batch silently fails to create a tag" still not fix yet

[JingsongLi]

• Batch automatic tagging still does not work: The coordinator commit in FlinkSink.doCoordinatorCommit directly discards the sink, bypassing BatchWriteGeneratorTagOperatorFactory; configuring tag.automatic-creation=batch does not create tags.
• end-input.watermark for streaming bounded-input is still not fully supported: EndInput events only enter the coordinator cache; in checkpoint-enabled mode, they are not committed immediately and still rely on the subsequent notifyCheckpointComplete call. When a bounded stream ends, there is typically no subsequent completed checkpoint. New tests now only assert the watermark in batch mode, while the streaming assertion has been bypassed.

[fishfishfishfishaa]

For batch automatic tag, I have added validation to prevent silent failures. I will complete the support for this in a follow-up, and it will be tracked later in #8220.
For end-input.watermark for streaming bounded-input, it is true that it relies on notifyCheckpointComplete to perform the commit. After the final EndInput, an empty committable is produced. Whether it creates a new snapshot depends on committer.forceCreatingSnapshot(). Otherwise, no new snapshot will be created, which means the end-input.watermark will not take effect. I have added tests for both the operator commit and coordinator commit which show consistent behavior.

@ifndef-SleePy PTAL and let me know if you have any suggestions. Thanks!

[ifndef-SleePy]

Hi @fishfishfishfishaa , sorry for late response.

I hive two design concerns about this PR:

1. It would be better to avoid blocking operations in the coordinator main thread, since this may block the JM main thread. (Update: Looks like your latest PR has already changed the implementation. The current approach looks more acceptable.)
2. It would be good to think through the failover cases carefully, for example task failover or global failover after some tasks have already sent end input.

A few small suggestions:

1. For complex logic like failover handling, it would be helpful to explain the design in a more structured way (maybe in the PR description or in comments), including how each failover case is handled. Otherwise, it is hard to tell from many if-else branches whether the code is fully correct.
2. It would be necessary to cover all cases with tests (especially the failover scenario), including both UTs and ITs.

[JingsongLi]

After all subtasks have issued END_INPUT, handleCommittableEvent performs the final commit only when checkpoints are disabled; when checkpoints are enabled, it must wait for the subsequent notifyCheckpointComplete. It is not guaranteed that another checkpoint will be completed before a bounded task ends, and close() only clears the event loop without committing cached END_INPUT committables. Therefore, data after the last checkpoint may remain uncommitted. Existing tests manually trigger subsequent checkpoints and do not cover scenarios where the task ends directly.

[fishfishfishfishaa]

Normal scenarios

Writers reach EndInput at different checkpoints

An early EndInput writer keeps its Long.MAX_VALUE entry. It is treated as already aligned for
later ordinary checkpoints, so the remaining writers can continue committing ordinary data. The
final entry is committed as soon as all writers reach EndInput. This final commit is triggered by
the last EndInput event and does not wait for notifyCheckpointComplete.

UT:

• CommittingWriteOperatorCoordinatorTest.testSubtasksEndInputAcrossDifferentCheckpoints

EndInput arrives after an already completed checkpoint

A bounded writer may produce its final data and EndInput only after the coordinator has already
processed notifyCheckpointComplete(N). The completed checkpoint is not reused as the final
commit trigger. When the last EndInput event arrives, the coordinator commits through
Long.MAX_VALUE directly and sets endInputCommitted=true.

This path does not require Flink to schedule another checkpoint after bounded tasks finish. In
particular, testLateEndInputAfterCheckpointCompletionIsCommitted uses periodic Flink checkpoints,
emits the final record only after the coordinator has observed a completed checkpoint, disables
checkpoints after tasks finish, and verifies the final result without manually triggering another
checkpoint.

UT&IT:

• CommittingWriteOperatorCoordinatorTest.testSubtasksEndInputAcrossDifferentCheckpoints
• CoordinatorEndInputCommitITCase.testCoordinatorEndInputIsCommittedWithoutCheckpointsAfterTasksFinish
• CoordinatorEndInputCommitITCase.testLateEndInputAfterCheckpointCompletionIsCommitted

Checkpoint completion after the final EndInput commit

If the last EndInput event is processed before a later checkpoint completion notification, it
performs the final commit and sets endInputCommitted=true. The same coordinator ignores later
EndInput events and checkpoint completion notifications, so the final commit is not repeated.

UT:

• CommittingWriteOperatorCoordinatorTest.testRepeatedEndInputAfterFinalCommitIsIgnored

Repeated EndInput before the final commit

EndInput may be replayed after recovery. The latest event is treated as the authoritative state for
that writer and replaces the previous Long.MAX_VALUE entry. It does not create a duplicate
ordinary checkpoint error.

UT:

• CommittingWriteOperatorCoordinatorTest.testRepeatedEndInputEventIsIdempotent
• WriterCommittablesTest.testRepeatedEndInputReplacesAuthoritativeEntry

Repeated EndInput after the final commit

When endInputCommitted=true, the same coordinator ignores later EndInput events and completed
checkpoint notifications.

UT:

• CommittingWriteOperatorCoordinatorTest.testRepeatedEndInputAfterFinalCommitIsIgnored

Failure scenarios

Checkpoint completion overtakes a delayed EndInput event

A writer may have already reached EndInput and persisted its Long.MAX_VALUE entry in operator
state while its CommittableEvent is still delayed before the coordinator. If completion of the
checkpoint containing that writer state is processed first, the coordinator may not cover the
completed checkpoint yet. In that case, alignCommittables returns false,
notifyCheckpointComplete throws Not all committables reported by writer, and the coordinator
event loop calls context.failJob.

The old attempt's delayed event is not used as recovery authority. After failover, each writer
restores from the completed checkpoint and sends one RestoredCommittableEvent. When the restored
events contain EndInput for all writers, the recovery target is Long.MAX_VALUE and the new
coordinator calls filterAndCommit. If this creates a missing snapshot, the configured recovery
behavior intentionally triggers one more restart so that writers restart from the new snapshot.

The IT controls checkpoint triggering only to reproduce this ordering deterministically. The
final data is committed from the restored EndInput committable, not by waiting for another
checkpoint completion.

UT&IT:

• CommittingWriteOperatorCoordinatorTest.testCheckpointCompletionBeforeEndInputEventTriggersFailover
• CoordinatorEndInputCommitITCase.testFinalCheckpointCompletionBeforeEndInputEventCommitsAfterFailover

WriteTask Failover

Writer task fails before EndInput

The coordinator remains in RUNNING. subtaskReset clears the failed writer's cached
committables. The recovered writer replays its pending state and then continues sending ordinary
checkpoint events.

UT:

• CommittingWriteOperatorCoordinatorTest.testPartialFailoverWithoutRestoring
• CommittingWriteOperatorCoordinatorTest.testPartialFailoverWithRestoring

A writer task fails after EndInput

subtaskReset clears the failed writer's cached EndInput entry. When the writer recovers, its
RestoredCommittableEvent may contain ordinary entries and a Long.MAX_VALUE entry.

While the coordinator is still RUNNING:

• already completed ordinary entries are ignored;
• the restored EndInput entry is saved again;
• it waits for the other writers; and
• the last EndInput event commits immediately, regardless of whether checkpointing is enabled.

UT:

• CommittingWriteOperatorCoordinatorTest.testPartialFailoverReplaysEndInputWhileCoordinatorIsRunning

All writers EndInput before another checkpoint completion

The EndInput entries are stored in writer operator state. After writer recovery, the restore events
rebuild allSubtasksEndInput(). When the last restored EndInput entry arrives in RUNNING, the
coordinator performs the final commit directly. It does not enter a state that waits for another
completed checkpoint.

UT:

• CommittingWriteOperatorCoordinatorTest.testPartialFailoverReplaysEndInputWhileCoordinatorIsRunning
• CommittingWriteOperatorCoordinatorTest.testPartialFailoverAfterAllSubtasksEndInputCommitted

JM Failover

JM failover with partial EndInput

The recreated coordinator enters RESTORING and waits for one restore event from every writer.
All writers restore from the same completed checkpoint.

When only some writers contain EndInput:

1. ordinary entries are aligned at the restored checkpoint;
2. already committed ordinary entries are filtered or cleared;
3. the Long.MAX_VALUE entries remain buffered;
4. the coordinator changes to RUNNING; and
5. it waits for the remaining writers to reach EndInput.

UT:

• CommittingWriteOperatorCoordinatorTest.testGlobalRestoreRetainsPartialEndInput

JM failover with all writers at EndInput

When all restore events contain EndInput, allSubtasksEndInput() becomes true during recovery. The
recovery target changes to Long.MAX_VALUE, and the coordinator calls filterAndCommit.

If recovery creates a missing snapshot, the configured recovery behavior triggers another job
restart so that all writers restart from the new snapshot. If the snapshot already exists,
filterAndCommit returns no new commit.

UT:

• CommittingWriteOperatorCoordinatorTest.testGlobalRestoreCommitsWhenAllSubtasksReachedEndInput

Batch Mode JM failover without a completed checkpoint

Without a completed checkpoint, there is no writer pending state to restore. Batch scheduling
recomputes the writer output. After all restarted writers send EndInput again, the coordinator
calls filterAndCommit(Long.MAX_VALUE).

UT&IT:

• CommittingWriteOperatorCoordinatorTest.testCheckpointDisabledCommitsWhenAllSubtasksEndInput

• CoordinatorEndInputCommitITCase.testCoordinatorCommitEndInputInBatchMode

Writer task and JM failover happen together or in sequence

JM failover subsumes a writer region failover that is still running or has just completed. In
Flink, each TaskExecutor observes the old JobMaster losing leadership, disconnects that
JobManager connection, and calls failExternally for every task of the job. The new JobMaster then
creates a new execution graph and globally restores every writer and the coordinator from the same
completed checkpoint. This does not fail the TaskManager process itself; it fails that job's tasks
running on the TaskManager. For each writer, restoring operator state is the same mechanism as an
ordinary task failover, but the JM failover scope is global and all writers use the same recovery
checkpoint. Therefore, a writer's local recovery attempt is not an additional recovery authority:

• The old coordinator's in-memory cache, local subtaskReset, and replay progress are discarded.
• If every checkpointed writer state contains EndInput, the restored coordinator aligns all writer
  events and uses Long.MAX_VALUE as the final recovery target.
• If only some checkpointed writer states contain EndInput, the restored coordinator recovers the
  ordinary checkpoint target and retains those Long.MAX_VALUE entries. They are committed only
  after the remaining writers reach EndInput; the last EndInput event triggers the final commit
  directly.

UT:

• CommittingWriteOperatorCoordinatorTest.testWriterFailoverThenGlobalRestoreUsesCheckpointedEndInput
• CommittingWriteOperatorCoordinatorTest.testWriterFailoverThenGlobalRestoreRetainsPartialCheckpointedEndInput

CK Abort

Checkpoint abort after partial EndInput

An aborted ordinary checkpoint does not remove pending committables. A later completed checkpoint
commits accumulated ordinary entries. The early Long.MAX_VALUE entry remains buffered until all
writers reach EndInput, then the last EndInput event commits it directly.

UT:

• CommittingWriteOperatorCoordinatorTest.testCheckpointAbortPreservesEarlyEndInput

[JingsongLi]

There is a race condition between END_INPUT and checkpoint completion. If the checkpoint completes first and END_INPUT is not received until later, the final event will not record and reuse this completed checkpoint; furthermore, a bounded job may stop generating checkpoints, resulting in the final batch of data never being committed. The current tests rely on manually triggering another checkpoint and do not cover the actual race condition.

[fishfishfishfishaa]

There is a race condition between END_INPUT and checkpoint completion. If the checkpoint completes first and END_INPUT is not received until later, the final event will not record and reuse this completed checkpoint; furthermore, a bounded job may stop generating checkpoints, resulting in the final batch of data never being committed. The current tests rely on manually triggering another checkpoint and do not cover the actual race condition.

Good catch. END_INPUT is sent through sendEventToCoordinator, while checkpoint completion is delivered through notifyCheckpointComplete. Although both ultimately enter the same JobMaster RpcEndpoint, the former uses an ask-style RPC and the latter a tell-style RPC. They therefore have different actor senders and do not satisfy Pekko’s same-sender FIFO condition, so their delivery order is not guaranteed.

Even with direct final commit after all END_INPUT events are collected, alignCommittables may still fail and trigger failover if checkpoint completion is processed before the END_INPUT event reaches the coordinator.

The current implementation addresses this in two ways:

1. In streaming mode, the coordinator performs an idempotent final commit immediately after collecting all END_INPUT events, without waiting for another checkpoint.
2. The writer persists pending END_INPUT committables in operator state. If the race triggers failover, the new coordinator restores and commits them through RestoredCommittableEvent, preventing data loss.

I added two ITs:

• testFinalCheckpointCompletionBeforeEndInputEventCommitsAfterFailover deterministically reproduces the strict race by controlling checkpoint timing and delaying the END_INPUT event. It verifies that a non-empty END_INPUT committable is restored and committed after failover. Manual checkpoints are used only to control ordering, not as a commit dependency.
• testLateEndInputAfterCheckpointCompletionIsCommitted uses periodic checkpoints and disables execution.checkpointing.checkpoints-after-tasks-finish. It verifies the final result and proves that the final batch is committed even when no later checkpoint is generated.

The current implementation ensures that the final commit is not lost. The race may trigger failover, but it does not affect consistency.

[ifndef-SleePy]

Before diving into implementation details, I'd like to discuss the design from a high-level view first.

Race condition issue:

The original CommitterOperator design:

• For batch jobs, it commits directly in endInput.
• For streaming with checkpointing enabled, it commits in notifyCheckpointComplete of the final checkpoint.

This PR instead always triggers the commit directly on end input.

For the streaming + checkpointing-enabled case, this differs from CommitterOperator, and I think the race condition the reviewer mentioned is not the low-level RPC ordering issue — it's that there are now two flows that can trigger a commit: end input and checkpoint completion. Although we handle them on a single thread and use the checkpoint id to distinguish them, and Paimon provides filterAndCommit to tolerate the overlap, this still feels weird: end input may commit the previous checkpoint's data ahead of time, before notifyCheckpointComplete is triggered, and that checkpoint is not even guaranteed to succeed. I'm not sure whether this has other side effects; it's a very subtle situation, and it gets more complex once we consider non-idempotent operators.

There is also the global failover case: the in-memory endInputCommitted variable is gone, so the behavior differs again — the coordinator will re-attempt both the checkpoint-driven and the end-input-driven commits. This should be filtered out by filterAndCommit and not actually re-commit, but it is again a new, different path.

Another case: if a writer calls endInput first and a normal checkpoint is triggered afterwards, the Long.MAX_VALUE end-input committable gets persisted into that writer's operator state. If a global failover then happens, on restore the writer reports a RestoredCommittableEvent with a normal (restore) checkpoint id, but its payload still carries the Long.MAX_VALUE committable. This feels odd, because the original design seems to expect that a restore only carries committables whose checkpoint id is less than or equal to the restore checkpoint id. I haven't verified whether this actually leads to an incorrect commit (the coordinator's headMap-based selection may exclude Long.MAX_VALUE anyway), but carrying it under a normal restore checkpoint id breaks the contract the code otherwise assumes.

Failover on finished task:

For a writer task that has already sent its end input, is there any scenario where it would not be brought up again to re-send RestoredCommittableEvent (e.g. deployed as FINISHED_ON_RESTORE)? My current understanding is that we should never reach FINISHED_ON_RESTORE in this scenario, so RestoredCommittableEvent is always re-sent, but I haven't verified this with a test. I'm not sure whether you've considered and tested this scenario.

[ifndef-SleePy]

Furthermore, the complexity of the current implementation comes from trying to support batch and streaming at the same time, and the fact that the data interaction and failover handling in coordinator commit is more complex than in operator commit.

• In batch and streaming-without-checkpoint, the commit should happen directly on end input, since there is no checkpoint
• in the streaming case, committing on checkpoint completion is more reasonable

PIP-30 is meant to solve the region failover problem in the streaming case, so we could discuss whether it's necessary to support batch and streaming-without-checkpoint at all.

• Introducing a coordinator in batch mode doesn't seem to bring any benefit — batch data can be persisted at shuffle boundaries, so there is no region failover problem.
• And streaming-without-checkpoint looks like a niche scenario that isn't worth the added complexity.

[ifndef-SleePy]

One more addition: the most important reason coordinator commit introduces end input is for testability. In most of PIP-30's scenarios, end input is never triggered, so increasing the complexity of the main logic just to support end input doesn't seem worthwhile. Based on our internal practice, without end input support it's fairly hard to implement the failover IT cases: while injecting random task failover and global failover, we need to verify that both the Flink and Paimon jobs recover normally and that data consistency holds against a fixed dataset.

[fishfishfishfishaa]

Thanks for pointing this out. The main motivation for supporting EndInput in coordinator commit was testability. However, the current implementation directly uses EndInput to avoid final commit loss, which indeed introduces additional complexity in both commit semantics and recovery.

I also agree that, for the PIP-30 goal of supporting region failover, batch mode and streaming without checkpointing provide limited practical value.

Based on this, I will try to simplify the implementation, narrow the supported scope, and remove the extra EndInput direct-commit path so that the commit and recovery semantics remain clearer.