hudi-agent commented on code in PR #20290:
URL: https://github.com/apache/hudi/pull/20290#discussion_r4253106466
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java:
##########
@@ -251,6 +276,21 @@ public void start() throws Exception {
.threadFactory(getThreadFactory("instant-request"))
.exceptionHook((errMsg, t) -> this.context.failJob(new
HoodieException(errMsg, t)))
.build();
+ if
(this.writeClient.getConfig().getPreWriteCleanerPolicy().isAsyncRollbackFailedWrites())
{
+ // No exception hook that fails the job: a rollback failure is logged
and retried on the next
+ // checkpoint, the same way the clean sink treats a failed clean.
+ // waitForTasksFinish: let an in-flight rollback drain on shutdown
instead of interrupting it
+ // mid-way, which would leave a pending rollback instant behind on
every restart. This matches
+ // how the clean sink treats the equivalent work.
+ this.rollbackExecutor = NonThrownExecutor.builder(log)
+ .threadFactory(getThreadFactory("prewrite-rollback"))
+ .waitForTasksFinish(true)
+ .build();
+ this.rollbackWriteClient = FlinkWriteClients.createWriteClient(conf);
+ // Roll back leftovers from a previous run right away rather than
waiting for the first
+ // checkpoint to complete; this is the case the pre-write rollback
exists for.
+ scheduleAsyncRollback();
Review Comment:
🤖 This startup rollback is submitted before `restoreEvents(Long.MAX_VALUE)`
(line 302). On restore, an instant c1 that was checkpointed but not committed
usually has an expired heartbeat by the time the job restarts;
`recommitInstant` calls `restartHeartbeat` for exactly this reason. That leaves
the rollback thread free to pick c1 and roll it back while `start()` is
recommitting it, which would drop data the restored checkpoint already
acknowledged. Could you move this call after `restoreEvents`?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java:
##########
@@ -251,6 +276,21 @@ public void start() throws Exception {
.threadFactory(getThreadFactory("instant-request"))
.exceptionHook((errMsg, t) -> this.context.failJob(new
HoodieException(errMsg, t)))
.build();
+ if
(this.writeClient.getConfig().getPreWriteCleanerPolicy().isAsyncRollbackFailedWrites())
{
Review Comment:
🤖 The safety argument depends on LAZY, but nothing enforces it, and Flink
COW upsert with a single writer stays on the default EAGER policy
(`FlinkWriteClients` only forces LAZY for non-blocking instant generation).
Under EAGER, `getInstantsToRollback` returns every pending instant. Here is the
interleaving: checkpoint N completes and the executor commits c1, then submits
the rollback; at t2 the instant-request thread starts c2; at t3 the rollback
thread lists pending instants and rolls back c2 while the tasks are still
writing to it. Could you enable this only when
`getFailedWritesCleanPolicy().isLazy()` (and fail fast otherwise), and also
check `areTableServicesEnabled()` the way the inline path does?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
