[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!