You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "vamshipasunuru1 (via GitHub)" <gi...@apache.org> on 2026/10/12 23:00:24 UTC
[PR] [UBER] Add ASYNC_ROLLBACK_FAILED_WRITES pre-write cleaner policy [hudi]
vamshipasunuru1 opened a new pull request, #20290:
URL: https://github.com/apache/hudi/pull/20290
### Summary
Today `hoodie.prewrite.cleaner.policy=ROLLBACK_FAILED_WRITES` rolls back failed writes inline inside `startCommit`, which runs on the Flink JobManager while a write task blocks on `Correspondent.requestInstantTime()`. Marker listing and file deletion therefore sit directly on the instant request path. Tuning timeouts is not reliable as any throttling from file-system can cause checkpoint failures .
This adds a policy that keeps the rollback on the JobManager but moves it off the critical path.
### Changelog
- HoodiePreWriteCleanerPolicy: new ASYNC_ROLLBACK_FAILED_WRITES value, with `isAsyncRollbackFailedWrites()` and `fromString` support for the lowercase form, matching how the existing values are parsed.
- BaseHoodieWriteClient#runPreWriteCleanerPolicy: the new policy is a logged no-op, so starting a commit never blocks on rollback. Engines that do not drive the rollback themselves perform no pre-write rollback under it.
- StreamWriteOperatorCoordinator: when the policy is set, builds a dedicated `prewrite-rollback` executor and write client, and submits a rollback at startup and after each completed checkpoint. Fire-and-forget with an in-flight guard, so neither the instant request path nor the checkpoint path ever waits; a skipped round is retried on the next checkpoint.
The rollback write client is deliberately separate from the coordinator's: the rollback runs through the table service client, whose TransactionManager tracks a single state-change owner, so driving it from two threads can leave the table lock held. This mirrors what CleanFunction already does.
The rollback client honours the configured `hoodie.rollback.parallelism` as-is; nothing is inferred or overridden. The rollback runs inside the JobManager JVM via HoodieFlinkEngineContext, so that value is the knob to tune if the default is too wide for the JobManager.
Safety relies on the LAZY failed-writes policy selecting instants by expired heartbeat, so a live writer's instant is never a candidate and running late is harmless.
### Impact
Reliable checkpointing without unbounded I/O work in the critical path.
### Risk Level
Low.
### Documentation Update
To enable this, set hoodie.prewrite.cleaner.policy=ASYNC_ROLLBACK_FAILED_WRITES. The value of this property is NONE by default.
### Contributor's checklist
- [ X] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ X] Enough context is provided in the sections above
- [ X] Adequate tests were added if applicable
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [UBER] Add async pre-write rollback support [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20290:
URL: https://github.com/apache/hudi/pull/20290#issuecomment-6134512534
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/20290?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `16.66667%` with `5 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 80.62%. Comparing base ([`a41f6ea`](https://app.codecov.io/gh/apache/hudi/commit/a41f6eaf7df0de8c9aefdd80a3cbb03967623e96?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`1376757`](https://app.codecov.io/gh/apache/hudi/commit/1376757077bfe857a6adf529af9afb2d1e257dd2?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/20290?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...hudi/common/model/HoodiePreWriteCleanerPolicy.java](https://app.codecov.io/gh/apache/hudi/pull/20290?src=pr&el=tree&filepath=hudi-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Fcommon%2Fmodel%2FHoodiePreWriteCleanerPolicy.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL21vZGVsL0hvb2RpZVByZVdyaXRlQ2xlYW5lclBvbGljeS5qYXZh) | 25.00% | [2 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/hudi/pull/20290?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../org/apache/hudi/client/BaseHoodieWriteClient.java](https://app.codecov.io/gh/apache/hudi/pull/20290?src=pr&el=tree&filepath=hudi-client%2Fhudi-client-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Fclient%2FBaseHoodieWriteClient.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1jbGllbnQvaHVkaS1jbGllbnQtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9odWRpL2NsaWVudC9CYXNlSG9vZGllV3JpdGVDbGllbnQuamF2YQ==) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/hudi/pull/20290?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #20290 +/- ##
============================================
- Coverage 80.62% 80.62% -0.01%
Complexity 35053 35053
============================================
Files 2552 2552
Lines 143402 143408 +6
Branches 17455 17456 +1
============================================
+ Hits 115620 115621 +1
- Misses 19858 19862 +4
- Partials 7924 7925 +1
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.14% <25.00%> (-0.01%)` | :arrow_down: |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.80% <0.00%> (-0.01%)` | :arrow_down: |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.97% <ø> (ø)` | |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `74.05% <ø> (ø)` | |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.19% <ø> (ø)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.80% <ø> (ø)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `71.11% <ø> (ø)` | |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.16% <ø> (ø)` | |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.52% <ø> (ø)` | |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.41% <ø> (ø)` | |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.00% <ø> (ø)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/20290/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `53.96% <ø> (ø)` | |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/20290/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/20290/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.96% <16.66%> (-0.01%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/20290?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...ache/hudi/sink/StreamWriteOperatorCoordinator.java](https://app.codecov.io/gh/apache/hudi/pull/20290?src=pr&el=tree&filepath=hudi-flink-datasource%2Fhudi-flink%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Fsink%2FStreamWriteOperatorCoordinator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1mbGluay1kYXRhc291cmNlL2h1ZGktZmxpbmsvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvc2luay9TdHJlYW1Xcml0ZU9wZXJhdG9yQ29vcmRpbmF0b3IuamF2YQ==) | `83.21% <ø> (ø)` | |
| [.../org/apache/hudi/client/BaseHoodieWriteClient.java](https://app.codecov.io/gh/apache/hudi/pull/20290?src=pr&el=tree&filepath=hudi-client%2Fhudi-client-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Fclient%2FBaseHoodieWriteClient.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1jbGllbnQvaHVkaS1jbGllbnQtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9odWRpL2NsaWVudC9CYXNlSG9vZGllV3JpdGVDbGllbnQuamF2YQ==) | `83.86% <0.00%> (-0.29%)` | :arrow_down: |
| [...hudi/common/model/HoodiePreWriteCleanerPolicy.java](https://app.codecov.io/gh/apache/hudi/pull/20290?src=pr&el=tree&filepath=hudi-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Fcommon%2Fmodel%2FHoodiePreWriteCleanerPolicy.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL21vZGVsL0hvb2RpZVByZVdyaXRlQ2xlYW5lclBvbGljeS5qYXZh) | `61.90% <25.00%> (-8.69%)` | :arrow_down: |
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [UBER] Add async pre-write rollback support [hudi]
Posted by "voonhous (via GitHub) via commits" <co...@hudi.apache.org>.
voonhous commented on PR #20290:
URL: https://github.com/apache/hudi/pull/20290#issuecomment-6701697955
Temporarily closing this PR while we debug a GitHub settings error on master (#20471); it will be reopened shortly, no action needed.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [UBER] Add async pre-write rollback support [hudi]
Posted by "voonhous (via GitHub) via commits" <co...@hudi.apache.org>.
voonhous closed pull request #20290: [UBER] Add async pre-write rollback support
URL: https://github.com/apache/hudi/pull/20290
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [UBER] Add async pre-write rollback support [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #20290:
URL: https://github.com/apache/hudi/pull/20290#discussion_r4254364273
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java:
##########
@@ -554,6 +629,51 @@ private String startInstant() {
return this.instant;
}
+ /**
+ * Submits a rollback of failed writes to the dedicated rollback executor, if one is configured.
+ *
+ * <p>This is deliberately fire-and-forget: nothing on the instant request path or the checkpoint
+ * path ever waits for it. When a rollback is still running, this round is skipped and the next
+ * checkpoint retries, so slow rollbacks cannot queue up behind each other.
+ *
+ * <p>Under the LAZY failed writes cleaning policy the instants picked up here are selected by
+ * expired heartbeat, so an instant that is still being written by a live writer is never a
+ * candidate and running late is harmless. For the same reason {@link #close()} abandons a running
+ * rollback instead of waiting for it.
+ */
+ private void scheduleAsyncRollback() {
+ if (rollbackExecutor == null || rollbackInFlight) {
+ return;
+ }
+ rollbackInFlight = true;
+ rollbackExecutor.execute(() -> {
+ try {
+ rollbackWriteClient.rollbackFailedWrites(rollbackWriteClient.getHoodieTable().getMetaClient());
Review Comment:
🤖 non-blocking: `resolveOrScheduleRollback` builds the marker-based rollback plan inside `txnManager.beginStateChange`, and Flink sets `FileSystemBasedLockProvider` whenever MDT is enabled. So it looks like `startCommit` on the instant-request thread still waits on that marker listing, through the shared table lock. Could the javadoc and description say so? Or, if throttled listing is the main case this targets, would it make sense to move plan scheduling out of the lock?
<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 +289,17 @@ 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:
🤖 It looks like this gate doesn't check `areTableServicesEnabled()`, so with `hoodie.table.services.enabled=false` the coordinator still rolls back failed writes, while `runPreWriteCleanerPolicy` skips the sync `ROLLBACK_FAILED_WRITES` policy in that case. Could you add the same guard here so the two policies behave the same?
<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:
##########
@@ -282,6 +331,17 @@ public void close() throws Exception {
if (hiveSyncExecutor != null) {
hiveSyncExecutor.close();
}
+ if (rollbackExecutor != null) {
+ // Deliberately no awaitTermination: close() runs on the JobManager main thread, so waiting for
+ // an in-flight rollback stalls every JobMaster RPC and can fail the application during a
+ // restart. Rollback is idempotent and marker based, so abandoning one is safe: whatever is
+ // left behind is picked up by the next coordinator that starts.
+ // shutdown() rather than shutdownNow(): an interrupt lands inside the listing and deletion
+ // the rollback is doing, where a ClosedByInterruptException aborts the cleanup half way and
+ // leaves the rollback heartbeat behind, blocking the retry until that heartbeat expires.
+ // There is nothing to gain by interrupting: the thread is a daemon and nothing waits on it.
+ rollbackExecutor.shutdown();
Review Comment:
🤖 Since the rollback keeps running after `close()`, could a new coordinator for the same table in the same JVM (session cluster, stop-with-savepoint then resubmit) pick up the same pending rollback `r1` while the old thread is still executing it? In single-writer mode there's no rollback heartbeat to prevent that. Both threads would then run `finishRollback`, which repeats the MDT update for `r1`. On timeline v2 it looks like it also writes a second completed `r1`, since only the inflight file's existence is checked. Is that overlap ruled out somewhere I'm missing? @danny0405
<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:
##########
@@ -554,6 +629,51 @@ private String startInstant() {
return this.instant;
}
+ /**
+ * Submits a rollback of failed writes to the dedicated rollback executor, if one is configured.
+ *
+ * <p>This is deliberately fire-and-forget: nothing on the instant request path or the checkpoint
+ * path ever waits for it. When a rollback is still running, this round is skipped and the next
+ * checkpoint retries, so slow rollbacks cannot queue up behind each other.
+ *
+ * <p>Under the LAZY failed writes cleaning policy the instants picked up here are selected by
+ * expired heartbeat, so an instant that is still being written by a live writer is never a
+ * candidate and running late is harmless. For the same reason {@link #close()} abandons a running
+ * rollback instead of waiting for it.
+ */
+ private void scheduleAsyncRollback() {
+ if (rollbackExecutor == null || rollbackInFlight) {
+ return;
+ }
+ rollbackInFlight = true;
+ rollbackExecutor.execute(() -> {
+ try {
+ rollbackWriteClient.rollbackFailedWrites(rollbackWriteClient.getHoodieTable().getMetaClient());
+ } catch (Throwable throwable) {
+ // The thread factory installs a fatal uncaught exception handler, so the task must not let
+ // anything escape: a failed rollback is retried on the next checkpoint instead.
+ log.error("Error rolling back failed writes for table {}",
+ conf.get(FlinkOptions.TABLE_NAME), throwable);
+ } finally {
+ rollbackInFlight = false;
+ if (rollbackExecutor.isShutdown()) {
+ // The coordinator closed while this rollback was running and left the client to us.
+ closeRollbackWriteClient();
Review Comment:
🤖 non-blocking: `closeRollbackWriteClient()` here is outside the `catch (Throwable)`, so anything thrown from `close()` reaches the thread's `FatalExitExceptionHandler` and exits the JobManager. One example is a class-loading failure after the user classloader is released at job end in a session cluster. Could you wrap it in its own try/catch?
<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()) {
+ // 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:
🤖 Moving it after `restoreEvents` only covers instants buffered in coordinator state; a pending instant restored from the write tasks' state is recommitted later in `handleBootstrapEvent`, after `start()` has returned, and its `restartHeartbeat` only runs once all bootstrap events arrive, so the start-time rollback could still pick that expired-heartbeat instant first. Would it make sense to drop the startup call and let the first `notifyCheckpointComplete` trigger the first rollback, since bootstrap recommits have finished by then?
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java:
##########
@@ -187,6 +191,40 @@ public class StreamWriteOperatorCoordinator
*/
private NonThrownExecutor hiveSyncExecutor;
+ /**
+ * A single-thread executor that rolls back failed writes off the instant request path.
+ *
+ * <p>Only instantiated when the pre-write cleaner policy is
+ * {@link HoodiePreWriteCleanerPolicy#ASYNC_ROLLBACK_FAILED_WRITES}, in which case the inline
+ * rollback inside {@code startCommit} is a no-op and this executor drives it instead.
+ *
+ * <p>A plain executor with a daemon thread rather than a {@link NonThrownExecutor}: closing the
+ * latter awaits termination, and {@link #close()} runs on the JobManager main thread, where any
+ * wait stalls every JobMaster RPC.
+ */
+ private ExecutorService rollbackExecutor;
+
+ /**
+ * A dedicated write client for the asynchronous rollback.
+ *
+ * <p>It must not be shared with {@link #writeClient}: the rollback runs through the table service
+ * client whose {@code TransactionManager} tracks a single state-change owner, so driving it from
+ * two threads can leave the table lock held after the losing thread ends its state change.
+ */
+ private HoodieFlinkWriteClient<?> rollbackWriteClient;
+
+ /**
+ * Guards {@link #rollbackWriteClient} against a double close: on shutdown the client is closed by
+ * whichever of {@link #close()} and the abandoned rollback task gets there last.
+ */
+ private final AtomicBoolean rollbackWriteClientClosed = new AtomicBoolean();
+
+ /**
+ * Flag that keeps at most one asynchronous rollback in flight; a skipped round is retried on the
+ * next checkpoint rather than queued.
+ */
+ private volatile boolean rollbackInFlight;
+
/**
Review Comment:
🤖 nit: could `rollbackInFlight` be an `AtomicBoolean` with `compareAndSet(false, true)` in `scheduleAsyncRollback`? That would match `rollbackWriteClientClosed` and make the single-flight intent explicit instead of a volatile check-then-set.
<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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [UBER] Add async pre-write rollback support [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org