[Feature][Connector-V2] Support two-phase commit exactly-once write for Hudi sink

[corgy-w]

Search before asking

• I had searched in the feature and found no similar feature requirement.

Description

The Hudi sink commits every flushed batch with the Hudi client auto-commit, so the records become visible before the checkpoint that contains them completes. When a job fails and the source replays the records of a checkpoint that never completed, those records are committed to Hudi a second time. The sink therefore only provides at-least-once delivery for Zeta, see docs/en/connectors/sink/Hudi.md and the earlier timer flush work in #11747.

It would be useful for the sink to support the two-phase commit protocol of the engine, so that the records of a checkpoint are committed to Hudi only after that checkpoint completes, which gives exactly-once delivery. Hudi itself supports this pattern: the data is written into an instant while the job is running, and the instant is committed afterwards, so readers only see completed instants.

Usage Scenario

• A streaming job replicates data from a CDC/JDBC source into Hudi and must not duplicate records when the job fails and restarts from a checkpoint.
• A batch job writes into Hudi and needs the whole checkpoint to be published atomically instead of batch by batch.

Related issues

• [Feature][Connector-V2] Add Hudi timer flush #11747 timer flush of the Hudi sink

[DanielLeens]

Classification: B / Connector-V2 Hudi exactly-once delivery.

Thanks for framing the failure window precisely. I rechecked current dev: HudiSinkWriter.prepareCommit() flushes the writer but returns no commit information, and HudiSink does not create a committer. That means checkpoint completion cannot currently own the visibility transition for a Hudi instant, so the at-least-once diagnosis is valid.

There is now one implementation path: #12440. Please keep further implementation and review on that PR rather than opening a competing change. The exact-once claim needs to be demonstrated at the checkpoint boundary: an instant must become visible only after the successful checkpoint completion callback; abort, task failure, and restore must not expose or duplicate records from an uncompleted checkpoint; and the commit/state serialization must preserve the recovery decision without changing existing default behavior.

The focused unit tests and Hudi E2E should include an induced failure/restart around the commit boundary, not just a successful checkpoint. The documentation should also state which delivery semantics apply when the new mode is not selected. Once #12440 has a complete, green CI result and those recovery cases are evidenced, it can receive a full merge review.