corgy-w commented on PR #12440:
URL: https://github.com/apache/seatunnel/pull/12440#issuecomment-6118530835
@DanielLeens thank you — Issues 1 and 2 were real, and the root cause was
the choice of the `LAZY` policy. Addressed in `54416149b6`.
## Issue 1 + 2 (High): the heartbeat no longer exists
Exactly-once now uses `HoodieFailedWritesCleaningPolicy.NEVER` instead of
`LAZY`. I checked each heartbeat-dependent branch in the Hudi 0.15.0 sources
you referenced:
* `BaseHoodieWriteClient.startCommit` → `if
(config.getFailedWritesCleanPolicy().isLazy()) {
heartbeatClient.start(instantTime); }` — with `NEVER` no heartbeat is started
at all, so the writer has no per-instant timer thread and no heartbeat file to
leak (Issue 2), including the file of an already committed instant.
* `HeartbeatUtils.abortIfHeartbeatExpired` (called by `commitStats`) → `if
(policy.isLazy() && heartbeatClient.isHeartbeatExpired(instant))` — with
`NEVER` the commit of a checkpointed instant cannot throw `Heartbeat ... has
expired`, no matter how long the writer has been gone (Issue 1, first half).
* `CleanerUtils.rollbackFailedWrites`: for the commit action it runs the
rollback only for `EAGER`; for the clean action it runs the lazy rollback
(`getInstantsToRollbackForLazyCleanPolicy`, i.e. inflight instants with an
expired or missing heartbeat) only for `LAZY`. With `NEVER` neither path can
roll the pending instant back (Issue 1, second half — the race you described).
The trade-off is explicit and documented: Hudi does not clean up failed
writes of an exactly-once table by itself, so the connector does it — the
record writer rolls back the instant of a checkpoint that never completed in
`close()`, the committer rolls it back in `abort()`, and an instant whose
checkpoint *did* complete is committed by `restoreCommit()`. An instance that
is killed hard enough that neither runs stays on the active timeline and has to
be rolled back manually; that is in the docs.
Regression tests, both of which fail under the old `LAZY` design:
*
`HudiTwoPhaseCommitTest#shouldCommitARestoredInstantWhenTheWriterIsGoneAndTheHeartbeatIsStale`
— writer closed, heartbeat file backdated by an hour, fresh committer (a "slow
restart"), `restoreCommit` still commits the instant.
*
`HudiTwoPhaseCommitTest#shouldNotStartAHeartbeatForAnInstantWaitingForItsCheckpoint`
— two checkpoints, and the heartbeat folder stays empty.
## Issue 3 (Medium): rejected records fail the checkpoint
`HudiRecordWriter#prepareCommit` now checks `hasErrors()`/`hasGlobalError()`
on the collected statuses and fails with `HudiConnectorException(HUDI-07)`,
reporting the number of rejected records, the first rejected key and its cause.
The checkpoint therefore never completes, so the records can be replayed
instead of being dropped from a checkpoint that is never replayed.
## Issue 4 (Medium): `WriteStatus` in the checkpoint state — partially
addressed
I did not move the state to `List<HoodieWriteStat>`, and I would like to
explain why before changing it: `HoodieJavaWriteClient#commit(instantTime,
writeStatuses, ...)` derives the stats with
`writeStatuses.stream().map(WriteStatus::getStat)` and then passes **both** the
stats and the statuses into `commitStats` → `commit(table, ..., stats,
writeStatuses)`. Committing with stats only therefore means calling the public
`commitStats(instantTime, HoodieListData.eager(emptyList()), stats, ...)` and
relying on the table commit not consuming the statuses.
`HoodieTableMetadataWriter#update` is metadata-driven in 0.15
(`update(HoodieCommitMetadata, HoodieData<HoodieRecord>, String)`) which is
reassuring, but I could not see the write-status consumer inside the
copy-on-write commit in the sources available to me, and the metadata table is
enabled by default (`hoodie.metadata.enable=true`), so I was not confident
enough to change the persistence format in a bug-fix commit.
What I did instead: with Issue 3 fixed, the `errors` map (and its
`Throwable`s) can no longer reach the checkpoint state, the `DefaultSerializer`
round trip is covered by
`HudiSinkAggregatedCommitterTest#shouldSurviveTheCheckpointSerialization`, and
the state-size trade-off is documented. I am happy to follow up with the
stats-only state — with a metadata-table-enabled test that proves the commit
stays consistent — if you agree it is safe, or as a separate change so this one
stays reviewable.
## Issue 5 (Medium): tests
(a) and (b) are the two regression tests above. (c), the fault-injection e2e
that kills the job between checkpoint completion and commit, is not in this
commit: it needs a way to fail a task at that exact point in the e2e container
(a flaky source or a container kill), so I would like to add it as a follow-up
unless you consider it a merge blocker. The current e2e plus the unit tests do
cover commit, restore and rollback against a real local table.
## Issues 6 and 7 (Low): docs
The end-of-job signal stays an ERROR log (now also on the committer side),
and it is documented. The "Exactly-once semantics" sections in `docs/en` and
`docs/zh` are rewritten: why `NEVER` is used and what it buys (time-independent
restore, no heartbeat files/threads), what the connector rolls back instead of
Hudi, the rejected-record behavior, the checkpoint-state size caveat, the "do
not downgrade a running job to `AT_LEAST_ONCE`" caveat, and the manual rollback
path for an orphaned instant.
Local verification on the new head (JDK 11): `spotless:apply` clean, `./mvnw
-o -pl seatunnel-connectors-v2/connector-hudi test -DforkCount=0` → BUILD
SUCCESS, **27 tests, 0 failures** (`HudiTwoPhaseCommitTest` 6/6). CI is running
on the new head and I will report the result here.
--
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]
