corgy-w opened a new pull request, #12440:
URL: https://github.com/apache/seatunnel/pull/12440
### Purpose of this pull request
- close #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
* [x] If any new Jar binary package adding in your PR, please add License
Notice according [New License
Guide](https://github.com/apache/seatunnel/blob/dev/docs/en/developer/new-license.md)
* [x] If necessary, please update the documentation to describe the new
feature.
* [x] 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`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
