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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
