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

[corgy-w]

Purpose of this pull request

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

The Hudi sink commits every flushed batch with the Hudi client auto-commit, so the records are published before the checkpoint that contains them completes, and a job that fails and replays the records of a not completed checkpoint commits them a second time.

This pull request adds an opt-in two-phase commit write path to the Hudi sink:

• A new semantics sink option with the values AT_LEAST_ONCE (default, unchanged behaviour) and EXACTLY_ONCE.
• With EXACTLY_ONCE the Hudi client auto-commit is disabled. All the batches written between two checkpoints are written into the same Hudi instant, and the instant is committed by an aggregated committer only after the checkpoint completes, so the readers never see the records of a checkpoint that was not completed.
• The commit info of a checkpoint (the instant time and the write statuses) is part of the checkpoint state. When a commit is lost, for example because the job failed between the checkpoint completion and the commit, restoreCommit applies it again after the restore. An instant that is already completed on the active timeline is skipped, so the records are never published twice.
• The instants of a checkpoint that never completed are not committed. The source replays those records after the pipeline restarts and Hudi rolls the orphan instants back when the writer heartbeat expires. Committing an instant that is not on the active timeline anymore fails the job with an explicit error instead of dropping data silently.
• The failed writes cleaning policy is set to LAZY for the exactly-once semantics: the default EAGER policy rolls back every inflight instant before a new instant is created, which would roll back the instants that are still waiting for the commit of their checkpoint.

Both Zeta and the Spark/Flink translations call SinkWriter#prepareCommit(long checkpointId) and commit through SinkAggregatedCommitter, which is what this implementation relies on.

Does this PR introduce any user-facing change?

Yes, it adds a user-facing option and documents it in docs/en/connectors/sink/Hudi.md and docs/zh/connectors/sink/Hudi.md:

• semantics = AT_LEAST_ONCE is the default and keeps the previous behaviour, so existing jobs are not affected.
• semantics = EXACTLY_ONCE requires the checkpoint to be enabled, and the records become visible only after a checkpoint completes, so the visibility latency is at least one checkpoint interval. With this semantics the timer flush and the batch_size flush only stage the records into the instant of the current checkpoint instead of publishing them. Parallel writers still follow the Hudi single writer concurrency control and should work on disjoint keys and partitions.

No incompatible change, docs/en/introduction/concepts/incompatible-changes.md is not affected.

How was this patch tested?

• HudiTwoPhaseCommitTest (new) runs the real writer and the real aggregated committer against a real Hudi table on the local file system and checks that
  • the two batches of one checkpoint are written with one instant and are invisible until the checkpoint commit,
  • the commit publishes exactly the expected number of records,
  • committing the same commit info twice (the lost commit recovery path) does not publish the records again,
  • the instant of an aborted checkpoint is rolled back and committing it afterwards fails instead of losing data silently,
  • the default AT_LEAST_ONCE semantics still commits every flushed batch with the client auto-commit.
• HudiSinkAggregatedCommitterTest (new) covers the commit info state serialization (it is part of the checkpoint state) and the writers that have nothing to commit.
• HudiSinkConfigTest (new) covers the new option and the failure on an unknown value.
• ./mvnw -pl seatunnel-connectors-v2/connector-hudi test passes all 24 tests of the module (JDK 11).
• ./mvnw -pl seatunnel-connectors-v2/connector-hudi,seatunnel-e2e/seatunnel-connector-v2-e2e/connector-hudi-e2e spotless:apply is applied and the e2e module compiles.
• A new e2e case HudiIT#testWriteHudiWithExactlyOnceSemantics with hudi/fake_to_hudi_exactly_once.conf writes 5 records with the exactly-once semantics and asserts that they are visible after the job finished. It runs in the CI container, the local run needs the e2e container images which were not available here.

Check list

• If any new Jar binary package adding in your PR, please add License Notice according New License Guide
• If necessary, please update the documentation to describe the new feature.
• If necessary, please update incompatible-changes.md to describe the incompatibility caused by this PR.
• If you are contributing the connector code, please check that the following files are updated: no new connector, so plugin-mapping.properties, seatunnel-dist/pom.xml, label-scope-conf.yml and config/plugin_config are unchanged; an e2e test case is added in seatunnel-e2e.

[DanielLeens]

What Problem Does This PR Solve?

• User pain point: The Hudi sink only ever wrote with Hudi's own client auto-commit. Every flushed batch became visible immediately and was committed independently of the engine's checkpoint, so a task restart after a partial failure could resubmit already-committed records and produce duplicate rows/commits. There was no way to get exactly-once delivery out of the Hudi sink, which the connector's own feature matrix openly admitted (exactly-once was unchecked).
• Fix approach: Introduce an opt-in semantics option (AT_LEAST_ONCE default / EXACTLY_ONCE). With EXACTLY_ONCE, the writer disables the Hudi client's auto-commit, accumulates all the batches of one checkpoint interval into a single, uncommitted Hudi instant, and hands the instant time plus write statuses to a new HudiSinkAggregatedCommitter via HudiCommitInfo/HudiAggregatedCommitInfo. The committer only commits (or, on Spark, rolls back) the instant once the engine's checkpoint containing it has completed, using the standard SeaTunnel two-phase-commit (SinkAggregatedCommitter) extension point. The commit is idempotent (skips instants already on the completed timeline) and the failed-writes cleaning policy is switched to LAZY so Hudi doesn't roll back the still-pending instant out from under the committer.
• One-sentence summary: Adds a genuine, checkpoint-gated two-phase-commit exactly-once mode to the Hudi sink, fully opt-in and backward compatible with the existing at-least-once behavior.

1. Code Change Review

1.1 Core Logic Analysis

Changed/added files (main):

• config/HudiSemantics.java (new enum: AT_LEAST_ONCE, EXACTLY_ONCE)
• config/HudiSinkConfig.java (+semantics field, +isExactlyOnce(), serialVersionUID bumped to 2L)
• config/HudiSinkOptions.java (+SEMANTICS option, default AT_LEAST_ONCE)
• exception/HudiErrorCode.java (+COMMIT_INSTANT_FAILED)
• sink/HudiSink.java (+createAggregatedCommitter(), +getAggregatedCommitInfoSerializer())
• sink/HudiSinkFactory.java (registers SEMANTICS as an optional option)
• sink/commit/HudiSinkAggregatedCommitter.java (new class: init/combine/commit/restoreCommit/abort/close)
• sink/state/HudiCommitInfo.java, sink/state/HudiAggregatedCommitInfo.java (new state/commit-info carriers)
• sink/writer/HudiRecordWriter.java (+exactlyOnce flag, instantTimeToWrite(), collectWriteStatuses(), prepareCommit(long), exactly-once-aware close())
• sink/writer/HudiSinkWriter.java (threads exactlyOnce into both writer-construction call sites, adds prepareCommit(long checkpointId))
• util/HudiUtil.java (withAutoCommit(!isExactlyOnce()), withFailedWritesCleaningPolicy(LAZY if exactlyOnce else EAGER))

Before/after, the crux of the change (HudiRecordWriter):

// before
private void executeWrite() {
    ...
    String writeInstantTime = writeClient.startCommit();
    switch (hudiTableConfig.getOpType()) {
        case INSERT: writeClient.insert(writeRecords, writeInstantTime); break;
        ...
    }
    writeRecords.clear();
}

// after
private String instantTimeToWrite(HoodieJavaWriteClient<HoodieAvroPayload> writeClient) {
    if (!exactlyOnce) {
        return writeClient.startCommit();
    }
    if (pendingInstantTime == null) {
        pendingInstantTime = writeClient.startCommit();
    }
    return pendingInstantTime;
}

With EXACTLY_ONCE, every batch written between two checkpoints reuses the same pendingInstantTime instead of starting a new instant per batch (HudiRecordWriter.java:235-243); the accumulated WriteStatuses are collected into pendingWriteStatuses (HudiRecordWriter.java:245-249) and only handed off in prepareCommit(checkpointId) (HudiRecordWriter.java:263-279), which is the sole path that clears pendingInstantTime/pendingWriteStatuses.

Key Findings:

1. The AT_LEAST_ONCE path is provably untouched in behavior: createAggregatedCommitter() returns Optional.empty() for that mode (HudiSink.java:110-117), which is exactly the default the SeaTunnelSink interface already returned before this PR (seatunnel-api/.../SeaTunnelSink.java:125-128), and HudiSinkWriter.prepareCommit still just flushes and reports nothing to commit when not exactly-once (HudiRecordWriter.java:263-267). Existing jobs that never set semantics get byte-for-byte identical runtime behavior.
2. withFailedWritesCleaningPolicy is deliberately switched to LAZY only for exactly-once (HudiUtil.java:198-212), with a comment explaining why: the default EAGER policy would roll back the very instant that is intentionally left inflight across a checkpoint boundary. This is the detail that makes the whole design work, and it's called out clearly.
3. HudiSinkAggregatedCommitter.commitInstant() re-reads the active timeline and treats an already-completed instant as a no-op (HudiSinkAggregatedCommitter.java:182-192), which is what makes commit()/restoreCommit() idempotent for the failure-then-retry case, and this is exercised by a real (non-mocked) integration-style test.
4. HudiSinkWriter.close()'s (via HudiRecordWriter.close()) exactly-once branch deliberately does not flush and commit any leftover buffered records that never reached a checkpoint boundary (HudiRecordWriter.java:311-338) — it only warns if an instant is still pending. That is correct: those records were never durably checkpointed, so the source is expected to replay them, and writing them into an instant nobody will ever commit would just leave permanent orphaned metadata behind.
5. HudiSinkAggregatedCommitter.init() is new code, and the SinkAggregatedCommitter interface explicitly documents that init() "will be called not once, each retry will call this" (seatunnel-api/.../SinkAggregatedCommitter.java:36-38). See Issue 2 below — a retry that calls init() again without an intervening close() silently replaces writeClientProvider/metaClient without closing the previous HoodieJavaWriteClient.

Runtime path (exactly-once), traced end to end:

SeaTunnelRow --> HudiSinkWriter.write()
             --> HudiRecordWriter.writeRecord() [buffers by HoodieKey in LinkedHashMap]
             --> (batch size reached OR timer FlushSignal on task thread)
             --> HudiRecordWriter.flush() [synchronized]
                 --> executeWrite()/executeDelete()
                     --> instantTimeToWrite(): reuses pendingInstantTime, only the FIRST
                         call in the interval does writeClient.startCommit()
                     --> writeClient.insert/upsert/bulkInsert/delete(records, pendingInstantTime)
                         [[withAutoCommit(false)]] -- instant stays REQUESTED/INFLIGHT
                     --> collectWriteStatuses(): pendingWriteStatuses.addAll(...)

--- checkpoint barrier reaches the sink task ---
HudiSinkWriter.prepareCommit(checkpointId)
    --> HudiRecordWriter.prepareCommit(checkpointId)
        --> flush() [drains any residual buffered batch]
        --> builds HudiCommitInfo(checkpointId, pendingInstantTime, copy of pendingWriteStatuses)
        --> clears pendingInstantTime / pendingWriteStatuses
    --> returned to the engine as this writer's per-checkpoint commit message

--- engine collects all writers' HudiCommitInfo for the checkpoint ---
HudiSinkAggregatedCommitter.combine(commitInfos)
    --> drops null / no-instant entries --> HudiAggregatedCommitInfo (this is the
        committer's CHECKPOINTED STATE)

--- checkpoint completes ---
HudiSinkAggregatedCommitter.commit(aggregatedCommitInfos)
    --> for each HudiCommitInfo: commitInstant()
        --> reloadActiveTimeline()
        --> if instant already completed: skip (idempotent)
        --> else if instant missing from active timeline: throw COMMIT_INSTANT_FAILED
        --> else: writeClient.commit(instantTime, writeStatuses, Option.empty())
            [instant becomes COMPLETED, records become visible]

--- on restart from a checkpoint whose commit may have been lost ---
HudiSinkAggregatedCommitter.restoreCommit(aggregatedCommitInfos) --> commit(...) [same idempotent path]

--- checkpoint aborted (Spark only, per SinkAggregatedCommitter contract) ---
HudiSinkAggregatedCommitter.abort(aggregatedCommitInfos)
    --> rollbackInstant(): rolls back the instant if not already completed

This does reach the real production write path, not just a recovery corner: every batch write in exactly-once mode goes through instantTimeToWrite()/collectWriteStatuses(), and every checkpoint completion goes through the new committer. This is not an isolated boundary-only change.

1.2 Compatibility Impact

Fully compatible.

• New option semantics defaults to AT_LEAST_ONCE, and that code path is unchanged in behavior (see Key Finding 1). Existing jobs with no semantics key parse and run identically.
• HudiSinkConfig bumped serialVersionUID from (implicit default before this change) to explicit 2L because a field was added to a Serializable class; this class is not part of any persisted checkpoint/savepoint state (it is only reconstructed from the job's own config at each run), so this does not affect restore-from-old-checkpoint compatibility.
• The new HudiCommitInfo/HudiAggregatedCommitInfo are brand new state/commit-info types with their own explicit serialVersionUIDs; there is no prior on-disk/checkpoint format to be compatible with since the aggregated committer for Hudi did not exist before this PR (createAggregatedCommitter() previously fell through to the interface default Optional.empty()).
• Multi-table sink support (SupportMultiTableSink, HudiSinkWriter implements SupportMultiTableSinkWriter<HudiClientManager>) pre-dates this PR and is verified intact: HudiSink still implements SupportMultiTableSink (HudiSink.java:53-57, unchanged by this diff), createWriter() still returns a SupportMultiTableSinkWriter (HudiSinkWriter.java:41-43, unchanged), and SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA is still in optionRule().optional(...) (HudiSinkFactory.java:72, pre-existing, only SEMANTICS was newly added on the line above it). The exactlyOnce flag is correctly threaded into both places a HudiRecordWriter is constructed, including the multi-table resource-manager re-initialization path (HudiSinkWriter.java:140-145), so multi-table sink usage is not silently downgraded to a different semantics than configured.
• No public API, protocol, or default value of any existing option changes.

1.3 Performance / Side-Effect Analysis

• CPU/memory: exactly-once mode buffers WriteStatus objects for the whole checkpoint interval in pendingWriteStatuses (HudiRecordWriter.java:96) instead of discarding them after each auto-committed batch. This is bounded by data volume per checkpoint interval and is the expected cost of 2PC — not a regression versus not supporting exactly-once at all, but worth calling out that a very long checkpoint interval combined with a very large per-interval volume will grow this list proportionally; that's inherent to the feature, not a bug.
• Networking/metadata IO: HudiSinkAggregatedCommitter.commitInstant() and rollbackInstant() each call reloadActiveTimeline() (HudiSinkAggregatedCommitter.java:184, :214), and both are invoked once per HudiCommitInfo inside the commit()/abort() loops (HudiSinkAggregatedCommitter.java:127-136, :164-173), i.e. once per parallel writer subtask, per checkpoint. For jobs with meaningful sink parallelism this means N timeline reloads (each a metadata read against the underlying filesystem, e.g. S3/HDFS) per checkpoint where a single reload per commit()/abort() call would suffice, since all commit infos in one call share the same table/timeline. This is a real, avoidable per-checkpoint IO cost that scales with parallelism (see Issue 3).
• Concurrency safety: instantTimeToWrite()/collectWriteStatuses() mutate plain (non-volatile, non-synchronized) pendingInstantTime/pendingWriteStatuses fields, called from executeWrite()/executeDelete() inside the pre-existing synchronized flush(), while prepareCommit() reads/clears them right after calling flush() but outside any lock (HudiRecordWriter.java:263-279). This mirrors the pre-existing (not introduced by this PR) synchronization pattern of write()/flush()/close() in this class, which relies on Zeta's single task-thread execution model (write, timer-flush and checkpoint-barrier processing are documented to run in order on the same task thread, per the writer's own comment at HudiSinkWriter.java:148-151). Given that pre-existing assumption already underlies this file's design, this PR does not introduce a new race beyond what already existed; flagged here only as a design note, not a new defect.
• Idempotency/retries: commit is correctly idempotent via the completed-instant check; rollback is correctly a no-op once already completed/absent (rollbackInstant, HudiSinkAggregatedCommitter.java:212-225).
• Resource release: HudiSinkAggregatedCommitter.close() correctly guards writeClientProvider != null before closing. HudiWriteClientProvider.close() already swallows/logs exceptions defensively (pre-existing). See Issue 2 for the one real resource-release gap this PR introduces (a re-init() on retry without closing the previous client).

1.4 Error Handling and Logging

Error handling is generally sound: commitInstant() fails loudly (COMMIT_INSTANT_FAILED) instead of silently skipping when an instant has disappeared from the active timeline, which is exactly the right choice for a data-integrity-sensitive path, and this is covered by a real test (HudiTwoPhaseCommitTest.shouldRollbackTheInstantOfAnAbortedCheckpoint, asserting the commit throws instead of losing data silently). No sensitive information (credentials, tokens) is logged anywhere in the diff.

Issue 1

• Location: seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/sink/writer/HudiRecordWriter.java:311-338 (close())
• Problem: When exactlyOnce is true and the writer is closed while an instant is still pending, the code only emits a LOG.warn and does not report this condition anywhere the engine or an operator would reliably see it (it's a per-writer-subtask log line, easy to miss in a large cluster). This is the visible symptom of configuring semantics = EXACTLY_ONCE without checkpointing actually enabled/completing: the job can run to completion and never make a single row visible in Hudi, discovered only by noticing a WARN log or an empty table.
• Potential risk: Silent "wrote nothing durable" outcome for an entire job run, discoverable only through log inspection, not through job status/metrics.
• Best improvement: Given the SinkWriter/SinkAggregatedCommitter API currently gives connectors no way to query whether checkpointing is enabled (confirmed: no such accessor exists on SinkWriter.Context or the sink API in this codebase), a hard fail-fast isn't feasible from inside this connector. A pragmatic improvement is to raise this to LOG.error (or route through the metrics/status-reporting path the engine already exposes for writer close, if any) so it is not lost among routine INFO/WARN noise, and to repeat the same warning in HudiSinkAggregatedCommitter.close()/restoreCommit() if a job is torn down with pending, never-committed aggregated state.
• Severity: Medium
• Raised by another reviewer: No

Issue 2

• Location: seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/sink/commit/HudiSinkAggregatedCommitter.java:88-105 (init())
• Problem: SinkAggregatedCommitter.init() is documented by the framework as being callable more than once on the same instance ("this method will be called not once. Each retry will call this", seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/SinkAggregatedCommitter.java:36-38). This PR's init() unconditionally assigns a brand new writeClientProvider and metaClient, without checking whether a previous writeClientProvider already holds an open HoodieJavaWriteClient and closing it first.
• Potential risk: On a retry that re-invokes init() without an intervening close(), the previous HoodieJavaWriteClient (and whatever internal resources it holds — executor/engine context, file handles) is dereferenced and leaked rather than closed.
• Best improvement: In init(), close the existing writeClientProvider if non-null before creating the replacement, mirroring the guard already used in close().
• Severity: Medium
• Raised by another reviewer: No

Issue 3

• Location: seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/sink/commit/HudiSinkAggregatedCommitter.java:124-136 and :163-173 (commit()/abort()), calling reloadActiveTimeline() inside the per-HudiCommitInfo loop
• Problem: The active timeline is reloaded from storage once per HudiCommitInfo rather than once per commit()/abort() invocation, even though every commit info processed in one call belongs to the same table/timeline.
• Potential risk: Unnecessary, parallelism-scaled metadata IO against the underlying filesystem (S3/HDFS/etc.) on every checkpoint completion; on tables with many parallel writer subtasks this multiplies timeline-listing calls without changing correctness.
• Best improvement: Reload the timeline once per commit(List<HudiAggregatedCommitInfo>)/abort(...) call (e.g., right after flattening all HudiCommitInfos) and reuse it for every instant check inside that call.
• Severity: Medium
• Raised by another reviewer: No

Issue 4

• Location: seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/sink/commit/HudiSinkAggregatedCommitter.java:156-162 (Javadoc on abort())
• Problem: The Javadoc states "Zeta never aborts a checkpoint without restarting the pipeline ... so this is the best-effort path for the other engines," which reads as if abort() plays an active role on Zeta/Flink. The SinkAggregatedCommitter interface itself documents abort() as being invoked "Only on Spark engine at now" (seatunnel-api/.../SinkAggregatedCommitter.java:65-71), i.e. it currently isn't called at all on Zeta or Flink.
• Potential risk: Low — this is a documentation-precision issue, not a functional one. The actual safety net for Zeta/Flink is correctly implemented elsewhere: the LAZY failed-writes-cleaning-policy plus Hudi's own writer-heartbeat expiry (HudiUtil.java:198-212), which the writer-side Javadoc describes correctly (HudiSinkWriter.java:105-113). A future reader could be misled into thinking abort() is doing real work on Zeta today when it currently is not exercised there.
• Best improvement: Tighten the abort() Javadoc to explicitly say it is currently invoked only by the Spark engine per the interface contract, and that Zeta/Flink recovery relies entirely on the heartbeat-based LAZY cleaning policy.
• Severity: Low
• Raised by another reviewer: No

2. Code Quality Assessment

2.1 Coding Standards

Every new core method (instantTimeToWrite, collectWriteStatuses, prepareCommit, combine, commit, restoreCommit, abort, commitInstant, rollbackInstant) and every new nontrivial field (exactlyOnce, pendingInstantTime, pendingWriteStatuses, HudiCommitInfo's three fields, HudiAggregatedCommitInfo.commitInfos) has an explanatory Javadoc or field comment describing purpose, lifecycle and constraints — this is well above the bar and clearly reflects a deliberate effort to make the exactly-once contract legible. No missing-comment issues found on the new code.

As a minor, pre-existing (not introduced by this PR) aside: the module already has an apparently unused, orphaned pair of classes at org.apache.seatunnel.connectors.seatunnel.hudi.state.HudiSinkState/HudiCommitInfo (old package, no serialVersionUID), distinct from the actively used org.apache.seatunnel.connectors.seatunnel.hudi.sink.state package this PR extends. Not touched by this diff and not a blocker here, but worth a follow-up cleanup since it could confuse future contributors about which HudiCommitInfo is live.

2.2 Test Coverage and Test Stability

Coverage is strong and, importantly, uses real assertions against a real (local, @TempDir-backed) Hudi table rather than mocks:

• HudiTwoPhaseCommitTest (new, 303 lines) verifies, against an actual HoodieTableMetaClient/timeline: (1) all batches of one checkpoint share a single instant and it stays uncommitted until the committer runs; (2) the committer's commit is idempotent (running it twice does not duplicate records or re-commit); (3) an aborted checkpoint's instant is rolled back and a subsequent commit attempt against it fails loudly instead of silently losing data; (4) the default AT_LEAST_ONCE semantics still commits every batch immediately (regression guard for Key Finding 1). Each of these assertions would legitimately fail if the corresponding production logic were broken (e.g., reverting instantTimeToWrite()'s reuse of pendingInstantTime would break the "one instant per checkpoint" assertion; reverting the completed-instant skip in commitInstant() would break the idempotency assertion and inflate totalRecords).
• HudiSinkAggregatedCommitterTest (new, 118 lines) verifies combine() correctly drops writers with nothing to commit, and that HudiAggregatedCommitInfo/HudiCommitInfo survive a real DefaultSerializer round trip (exercises the actual checkpoint-state serialization path, not a stand-in).
• HudiSinkConfigTest (new, 70 lines) verifies the default semantics, explicit EXACTLY_ONCE parsing, and that an unknown enum value fails config parsing.
• New e2e testWriteHudiWithExactlyOnceSemantics in HudiIT.java runs a real batch job with checkpoint.interval = 1000 against the exactly-once sink and asserts the expected row count via ParquetReader after the job completes.

Stability rating: Stable, with one caveat carried over from the file's pre-existing convention rather than newly introduced by this PR — testWriteHudiWithExactlyOnceSemantics uses given().ignoreExceptions().await().atMost(60000, ...) (HudiIT.java:158-174), which is a weak-readiness pattern (swallowing exceptions while polling) in the abstract, but it is exactly the same pattern already used by the two pre-existing test methods in this same file (testWriteHudi, testWriteHudiWithOmitConfigItem), so this PR does not lower the bar versus what already exists here; it's Low severity and consistent with established local convention, not a new regression risk. HudiTwoPhaseCommitTest correctly uses @DisabledOnOs(OS.WINDOWS) for the same reason the rest of the Hudi test suite already skips Windows (Hadoop/native-path issues), matching the CI evidence in this PR's own comments that connector-hudi unit tests (including the skipped-on-Windows two-phase-commit test) pass in every attempted CI run. No hard sleeps, no floating-point-without-delta, no order-dependence, and resources (writer.close()/committer.close()) are released in finally blocks throughout the new tests.

2.3 Documentation Updates

Both docs/en/connectors/sink/Hudi.md and docs/zh/connectors/sink/Hudi.md are updated and consistent with each other and with the implementation: the feature matrix checkbox for exactly-once is flipped, the semantics option is documented in the options table and its own ### semantics [Enum] section, and a new "Exactly-once semantics" section walks through the mechanism (how it works, requirements/trade-offs, including the correct call-out that checkpointing must be enabled and that visibility latency becomes at least one checkpoint interval, and the single-writer-per-table concurrency caveat inherited from Hudi itself). This is genuinely good documentation — it explains the why, not just the what, and both languages say the same thing.

3. Architectural Soundness

3.1 Elegance of the Solution

Long-term solution. This is not a workaround: it reuses the framework's existing SinkAggregatedCommitter/CommitInfoT/AggregatedCommitInfoT extension point exactly as other two-phase-commit sinks in this codebase do, and it leans on Hudi's own instant/timeline/heartbeat machinery (auto-commit toggle, failed-writes cleaning policy) rather than inventing a parallel bookkeeping mechanism. The design correctly separates "write into an instant" (writer-side, in-memory, replayable) from "commit the instant" (committer-side, checkpoint-gated, idempotent, restorable), which is the right shape for this problem.

3.2 Maintainability

Good. The new classes are small, single-purpose, and heavily commented on intent rather than mechanics. The two Medium issues above (retry-time resource leak in init(), per-commit-info timeline reload) are both narrow, local fixes that don't require restructuring anything.

3.3 Extensibility

The HudiSemantics enum and the isExactlyOnce() boolean threaded through HudiSinkConfig/HudiRecordWriter/HudiUtil give a clean seam if a third semantics mode were ever needed, without touching the at-least-once path.

3.4 Historical-Version Compatibility

No impact on historical jobs, checkpoints, or savepoints: the new committer, state, and commit-info types did not exist before this PR (there was no aggregated committer for Hudi at all), so there is nothing for them to be incompatible with, and the at-least-once path — which is what every pre-existing job configuration uses by default — is unchanged in behavior.

4. Issue Summary

Number	Issue	Location	Severity
1	close() silently warns (log-only) instead of surfacing that an exactly-once job produced no committed output; no fail-fast possible given current API	HudiRecordWriter.java:311-338	Medium
2	init() can be called again on retry per the SinkAggregatedCommitter contract, but doesn't close a previously-created writeClientProvider/client first	HudiSinkAggregatedCommitter.java:88-105	Medium
3	Active timeline reloaded once per HudiCommitInfo instead of once per commit()/abort() call, scaling metadata IO with parallelism	HudiSinkAggregatedCommitter.java:124-136, :163-173	Medium
4	abort() Javadoc overstates its role on Zeta/Flink versus the interface's documented Spark-only invocation	HudiSinkAggregatedCommitter.java:156-162	Low

5. Merge Recommendation

Conclusion: Ready to merge after fixes

There are no prior reviews on this PR to agree or disagree with (this is the first formal review). CI is currently red, but the failure is conclusively unrelated to this change: the only failing lane is unit-test (11, windows-latest) (apache-side "Build" check is just a pointer to the actual run in the contributor's fork), and it fails in connector-http-paypal's PayPalClientTest#closeWakesRetryWait — a pre-existing, already-tracked timing flake (#12381), reproduced identically across four separate reruns, in a module this PR does not touch. I independently confirmed from the repository history that origin/dev already contains the prior mitigation for this flake (#12370, "widen PayPalClientTest's loopback HTTP wait budget") at a commit that predates this PR's base, and that #12370 is not sufficient — the actual root-cause fix (#12444) is still open/unmerged. Syncing this branch to latest dev would not fix this failure since the real fix hasn't landed anywhere yet. connector-hudi's own unit tests and the new connector-hudi-e2e case (including the new exactly-once test) pass in every attempt of this CI run. This CI failure is not a blocker.

1. Blockers — must be fixed (sorted by severity, CI-only findings excluded per policy): none of the findings above are release-blocking data-correctness bugs — the core two-phase-commit mechanism (instant reuse, checkpoint-gated commit, idempotency, rollback-then-fail-fast) is correct and is backed by real, failure-capable tests. However, given this is a durability feature, I'd still like to see Issues 1 and 2 addressed before merge, since they touch exactly the failure/retry paths this feature exists to make safe:

   • Issue 2 (Medium): guard init() against leaking a previous HoodieJavaWriteClient on retry — small, local fix.
   • Issue 1 (Medium): elevate the "job finished without committing anything" signal above a routine WARN log, since there's currently no other way an operator would notice a misconfigured (EXACTLY_ONCE + checkpoint disabled) job silently wrote nothing durable.

2. Recommended fixes — non-blocking:

   • Issue 3 (Medium, but non-blocking): reload the active timeline once per commit()/abort() call instead of per commit info — a straightforward efficiency improvement, not a correctness bug, and unlikely to matter until sink parallelism is large.
   • Issue 4 (Low): tighten the abort() Javadoc to match the interface's documented Spark-only invocation, since the actual Zeta/Flink safety net (heartbeat-based LAZY cleaning policy) is already correct and separately, correctly documented.

Overall assessment: This is a well-designed, well-tested addition that finally gives the Hudi sink a real exactly-once mode built on the framework's standard two-phase-commit extension point rather than a bespoke mechanism, and it does not touch or risk the existing at-least-once behavior at all. The Medium findings are narrow and local (a missing close-before-replace guard, and a log-visibility gap), not architectural problems. I don't see a meaningfully different alternative implementation worth trading off here — reusing SinkAggregatedCommitter plus Hudi's own instant/heartbeat machinery is the natural fit, and the PR does exactly that.