hudi-agent commented on code in PR #19960:
URL: https://github.com/apache/hudi/pull/19960#discussion_r4020902863
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java:
##########
@@ -412,22 +449,76 @@ public CompletableFuture<CoordinationResponse>
handleCoordinationRequest(Coordin
}
private CompletableFuture<CoordinationResponse>
handleInstantRequest(Correspondent.InstantTimeRequest request) {
- CompletableFuture<CoordinationResponse> response = new
CompletableFuture<>();
- instantRequestExecutor.execute(() -> {
- long checkpointId = request.getCheckpointId();
- Pair<String, EventBuffer> instantTimeAndEventBuffer =
this.eventBuffers.getInstantAndEventBuffer(checkpointId);
- final String instantTime;
- if (instantTimeAndEventBuffer == null) {
- // wait until previous instants are committed.
- eventBuffers.awaitAllInstantsToCompleteIfNecessary();
- instantTime = startInstant();
- this.eventBuffers.initNewEventBuffer(checkpointId, instantTime);
- } else {
- instantTime = instantTimeAndEventBuffer.getLeft();
+ final long checkpointId = request.getCheckpointId();
+ if (isClosing) {
+ return failedResponse("Coordinator is closing, rejecting instant request
for checkpoint " + checkpointId);
+ }
+ // Idempotent fast path: the checkpoint -> instant mapping is
authoritative and survives op retirement,
+ // so a lost READY reply is recovered by the next poll without creating a
second instant.
+ Pair<String, EventBuffer> instantTimeAndEventBuffer =
this.eventBuffers.getInstantAndEventBuffer(checkpointId);
+ if (instantTimeAndEventBuffer != null) {
+ return readyResponse(instantTimeAndEventBuffer.getLeft());
+ }
+ // Register or join exactly one creation for this checkpoint; only the
first caller submits the work.
+ final long currentEpoch = this.epoch;
+ InstantOp op = instantOps.computeIfAbsent(checkpointId, cid -> new
InstantOp(cid, currentEpoch));
+ if (op.claimCreation()) {
+ NonThrownExecutor worker = this.instantRequestExecutor;
+ if (worker != null && !isClosing) {
+ worker.execute(() -> runInstantCreation(op), "create instant for
checkpoint %d", checkpointId);
}
-
response.complete(CoordinationResponseSerDe.wrap(Correspondent.InstantTimeResponse.getInstance(instantTime)));
- }, "request instant time");
- return response;
+ }
+ // Reply with the current status without blocking; the requester polls
until READY/FAILED.
+ switch (op.getStatus()) {
+ case READY:
+ return readyResponse(op.getInstant());
+ case FAILED:
+ return failedResponse(op.getError());
+ default:
+ return CompletableFuture.completedFuture(
+
CoordinationResponseSerDe.wrap(Correspondent.InstantTimeResponse.pending()));
+ }
+ }
+
+ /**
+ * Creates a new instant for the given operation on the instant-request
worker thread.
+ *
+ * <p>Runs off the coordination-RPC path so the RPC can always answer in
O(1). The result is only
+ * published (buffer installed, status flipped to READY) when the
coordinator generation is unchanged,
+ * so a concurrent failover/close never observes a stale instant. Any
failure is terminal: the op is
+ * marked FAILED for prompt requester feedback and the exception is rethrown
so the executor's exception
+ * hook fails the job.
+ */
+ private void runInstantCreation(InstantOp op) {
+ if (op.getEpoch() != this.epoch || isClosing) {
+ // fenced before starting: a newer generation (or a close) took over.
+ return;
+ }
+ try {
+ // ordering: wait until all prior-checkpoint instants are committed
(blocking-generation mode only).
+ eventBuffers.awaitAllInstantsToCompleteIfNecessary(op.getCheckpointId());
+ String instantTime = startInstant();
+ if (op.getEpoch() != this.epoch || isClosing) {
+ // fenced after starting: do not publish the now-stale instant;
recovery reconciles the
+ // requested instant left on the timeline via recommit/rollback.
+ return;
Review Comment:
🤖 When we get fenced here, `startInstant()` has already run
`heartbeatClient.start(instant)` (LAZY policy, which `FlinkWriteClients` forces
for non-blocking-generation tables) and nothing ever calls
`cleanResources(instant)` for it, so the lazy cleaner sees a live heartbeat and
never rolls it back while this coordinator lives — it just stays inflight and
pins archival / cleaner retention. Could we call
`writeClient.cleanResources(instantTime)` before returning (or explicitly roll
it back) so the fenced instant actually gets reclaimed? Note the old code would
have published it and reused it on the next request for the same checkpoint, so
this is a new leak in the global-failover-during-slow-startInstant case.
<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:
##########
@@ -265,11 +293,16 @@ public void start() throws Exception {
@Override
public void close() throws Exception {
+ // reject new instant requests and fence any in-flight creation from
publishing.
+ this.isClosing = true;
+ this.epoch++;
// teardown the resource
if (executor != null) {
executor.close();
}
if (instantRequestExecutor != null) {
+ // waitForTasksFinish(true) drains a running instant creation (e.g.
inside startInstant()) before
+ // the write client is closed below.
instantRequestExecutor.close();
Review Comment:
🤖 With `waitForTasksFinish(true)` this is now `shutdown()` +
`awaitTermination(MAX)` instead of the previous `shutdownNow()`, and by this
point `executor` has already been drained so no `reset()`/`unblock()` can ever
arrive. If the worker is sitting in `CommitGuard.blockFor` (blocking-generation
mode, waiting for the prior checkpoint's commit that will never come once the
job is cancelling), `close()` blocks the JobMaster main thread for the full
`write.commit.ack.timeout` (= checkpoint timeout, 10 min by default). Since the
epoch/`isClosing` fence already prevents a stale publish, could we keep the
interrupting shutdown here, or signal the commit guard from `close()` before
waiting? @danny0405 wdyt on the intended close semantics here?
<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:
##########
@@ -756,6 +856,57 @@ public OperatorCoordinator create(Context context) {
/**
* Remember some table state variables.
*/
+ /**
+ * State of an asynchronous instant creation for one checkpoint.
+ *
+ * <p>Created on the coordinator thread when the first request for a
checkpoint arrives and mutated by
+ * the {@link #instantRequestExecutor} worker as the creation progresses.
{@code status}/{@code instant}/
+ * {@code error} are read back by later requests on the coordinator thread,
hence {@code volatile}. The
+ * {@code epoch} pins the creation to the coordinator generation that
submitted it so a stale result is
+ * never published after a failover or close.
+ */
+ private static class InstantOp {
+ final long checkpointId;
+ final long epoch;
+ private final AtomicBoolean creationClaimed = new AtomicBoolean(false);
+ @Getter
+ private volatile Correspondent.Status status =
Correspondent.Status.PENDING;
+ @Getter
+ private volatile String instant;
+ @Getter
+ private volatile String error;
+
+ InstantOp(long checkpointId, long epoch) {
+ this.checkpointId = checkpointId;
+ this.epoch = epoch;
Review Comment:
🤖 nit: `status`/`instant`/`error` use lombok `@Getter` but
`checkpointId`/`epoch` get hand-written getters right below — could you
annotate all the fields consistently (either all `@Getter` or all manual) so
it's not two styles side by side?
<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/event/Correspondent.java:
##########
@@ -121,10 +207,36 @@ public static InstantTimeRequest getInstance(long
checkpointId) {
@Getter
public static class InstantTimeResponse implements CoordinationResponse {
+ private final Status status;
private final String instant;
+ private final String errorMessage;
+
+ /**
+ * The instant is ready to use.
+ */
+ public static InstantTimeResponse ready(String instant) {
+ return new InstantTimeResponse(Status.READY, instant, null);
+ }
+
+ /**
+ * The instant is still being created, the requester should poll again.
+ */
+ public static InstantTimeResponse pending() {
+ return new InstantTimeResponse(Status.PENDING, null, null);
+ }
+
+ /**
+ * The instant creation failed terminally.
+ */
+ public static InstantTimeResponse failed(String errorMessage) {
+ return new InstantTimeResponse(Status.FAILED, null, errorMessage);
+ }
+ /**
+ * Kept for backward compatibility: a ready response carrying the instant.
Review Comment:
🤖 nit: "Kept for backward compatibility" is a bit misleading here —
`getInstance` isn't preserving a serialized/wire format, it's just a factory
alias for `ready(...)`. Might be clearer to say it's a convenience alias, or
just remove it if nothing outside tests still calls it.
<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]
