hudi-agent commented on code in PR #19960:
URL: https://github.com/apache/hudi/pull/19960#discussion_r4056698348
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java:
##########
@@ -134,13 +141,22 @@ public HashMap<Long, String>
getAllCheckpointIdAndInstants() {
}
public void initNewEventBuffer(long checkpointId, String instantTime) {
- this.eventBuffers.put(checkpointId, Pair.of(instantTime, new
EventBuffer(dataWriteParallelism, indexWriteParallelism)));
+ this.eventBuffers.compute(checkpointId, (cid, existing) -> {
+ ValidationUtils.checkState(existing == null,
+ String.format("Checkpoint %d is already bound to instant %s,
refusing to rebind to %s",
+ cid, existing == null ? null : existing.getLeft(), instantTime));
+ return Pair.of(instantTime, new EventBuffer(dataWriteParallelism,
indexWriteParallelism));
+ });
}
- public void awaitAllInstantsToCompleteIfNecessary() {
- if (this.commitGuardOption.isPresent() && nonEmpty()) {
- this.commitGuardOption.get().blockFor(getPendingInstants());
- }
+ /**
+ * Waits until all the pending instants before {@code checkpointId} are
committed.
+ *
+ * <p>Only blocks in blocking-instant-generation mode (a commit guard is
present); otherwise no-op.
+ * Uses a predicate loop so a spurious wakeup does not let a new instant
start before prior commits finish.
+ */
+ public void awaitAllInstantsToCompleteIfNecessary(long checkpointId) {
+ this.commitGuardOption.ifPresent(guard -> guard.blockFor(() ->
getPendingInstantsBefore(checkpointId)));
Review Comment:
🤖 This switches the guard predicate from `nonEmpty()` (any buffered events)
to map keys, which is more robust in the normal flow — but it also means an
entry with no events now blocks. A `createInstant(X)` that survives a regional
failover and publishes an empty entry X *after* X+1 has already been committed
can only be swept by `commitInstants(X+2)`, which the writers blocked here on X
can't reach. Could that case be handled (e.g. skip/sweep entries with no
events, or drop late creations), or is the resulting ack-timeout stall
considered acceptable?
<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:
##########
@@ -412,22 +425,52 @@ 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();
- }
-
response.complete(CoordinationResponseSerDe.wrap(Correspondent.InstantTimeResponse.getInstance(instantTime)));
- }, "request instant time");
- return response;
+ final long checkpointId = request.getCheckpointId();
+ // Idempotent fast path: the checkpoint -> instant mapping is
authoritative and survives marker 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());
+ }
+ // Atomically submit exactly one creation for this checkpoint. Later polls
only inspect state.
+ if (instantCreationCheckpoints.add(checkpointId)) {
+ this.instantRequestExecutor.execute(
+ () -> createInstant(checkpointId), "create instant for checkpoint
%d", checkpointId);
+ }
+
+ // A synchronous test executor may have completed creation already;
production workers normally return PENDING here.
+ instantTimeAndEventBuffer =
this.eventBuffers.getInstantAndEventBuffer(checkpointId);
+ if (instantTimeAndEventBuffer != null) {
+ return readyResponse(instantTimeAndEventBuffer.getLeft());
+ }
+ return CompletableFuture.completedFuture(
+
CoordinationResponseSerDe.wrap(Correspondent.InstantTimeResponse.pending()));
+ }
+
+ /**
+ * Creates a new instant for the given checkpoint on the instant-request
worker thread.
+ *
+ * <p>Runs off the coordination-RPC path so the RPC can always answer in
O(1). Any failure is propagated
+ * through the executor's existing exception hook, which fails the job
through the normal failure path.
+ */
+ private void createInstant(long checkpointId) {
+ // A recovery may restore the mapping after the task was submitted but
before it starts.
+ if (this.eventBuffers.getInstantAndEventBuffer(checkpointId) != null) {
+ return;
+ }
+
+ // Ordering: wait until all prior-checkpoint instants are committed
(blocking-generation mode only).
+ this.eventBuffers.awaitAllInstantsToCompleteIfNecessary(checkpointId);
Review Comment:
🤖 In blocking-generation mode (COW/CDC upsert), if checkpoint K is aborted
after the writers snapshotted, the writers request K from inside
`processElement`/`snapshotState` and this guard waits for K-1 to commit — but
K-1 only commits on the *next* checkpoint completing, which the polling writers
can't process. Previously the 10s ask timeout broke that circular wait quickly;
now it holds for the full `write.commit.ack.timeout` (300s) before `failJob`.
Is a 5-minute stall on a checkpoint abort acceptable here, or should something
like `notifyCheckpointAborted` release the guard?
<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:
##########
@@ -412,22 +425,52 @@ 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();
- }
-
response.complete(CoordinationResponseSerDe.wrap(Correspondent.InstantTimeResponse.getInstance(instantTime)));
- }, "request instant time");
- return response;
+ final long checkpointId = request.getCheckpointId();
Review Comment:
🤖 nit: `handleInstantRequest` now does the lookup/submit/re-lookup dance
inline — worth double-checking whether the duplicated
`getInstantAndEventBuffer` + `readyResponse` calls before/after the submit
could be collapsed into a small loop or helper to make the fast-path vs.
race-path handling more obvious at a glance.
<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:
##########
@@ -64,16 +90,69 @@ public static Correspondent getInstance(OperatorID
operatorID, TaskOperatorEvent
}
/**
- * Sends a request to the coordinator to fetch the instant time.
+ * Requests the instant time for the given checkpoint from the coordinator.
+ *
+ * <p>The coordinator answers each request in O(1) with a {@link Status}:
the requester polls with a
+ * capped exponential backoff (plus jitter) under a single {@code
pollBudgetMs} deadline until the
+ * instant is {@code READY}, and retries transient transport errors within
the same budget. A
+ * {@code PENDING} reply never extends the deadline. Instant creation
failures fail the job through
+ * the coordinator's normal asynchronous failure path.
+ *
+ * @param checkpointId The checkpoint id (or -1 for bulk insert)
+ * @param pollBudgetMs The overall budget to wait for an instant, in
milliseconds
+ *
+ * @return the instant time to write with
*/
- public String requestInstantTime(long checkpointId) {
+ public String requestInstantTime(long checkpointId, long pollBudgetMs) {
+ final long deadlineNanos = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(pollBudgetMs);
+ long backoffMs = POLL_BASE_MS;
+ while (true) {
+ InstantTimeResponse response;
+ try {
+ response = fetchInstantTimeResponse(checkpointId);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new HoodieException("Interrupted while requesting the instant
time from the coordinator", e);
+ } catch (Exception e) {
+ // transient transport/coordinator error: retry within the budget,
reusing the same checkpoint identity.
+ if (System.nanoTime() >= deadlineNanos) {
+ throw new HoodieException("Timeout requesting the instant time from
the coordinator for checkpoint " + checkpointId, e);
+ }
+ backoffMs = sleepAndGrow(backoffMs);
+ continue;
+ }
+ if (response.getStatus() == Status.READY) {
+ return response.getInstant();
+ }
+ // PENDING: keep polling, but never reset the deadline.
+ if (System.nanoTime() >= deadlineNanos) {
+ throw new HoodieException("Timeout waiting for the instant time from
the coordinator for checkpoint " + checkpointId);
+ }
+ backoffMs = sleepAndGrow(backoffMs);
+ }
+ }
+
+ /**
+ * Sends a single instant-time request to the coordinator and returns its
response.
+ *
+ * <p>Isolated so tests can stub the transport while reusing the poll loop
in {@link #requestInstantTime}.
+ */
+ protected InstantTimeResponse fetchInstantTimeResponse(long checkpointId)
throws Exception {
+ return
CoordinationResponseSerDe.unwrap(this.gateway.sendRequestToCoordinator(this.operatorID,
+ new
SerializedValue<>(InstantTimeRequest.getInstance(checkpointId))).get());
+ }
+
+ private static long sleepAndGrow(long backoffMs) {
+ long capped = Math.min(backoffMs, POLL_CAP_MS);
Review Comment:
🤖 nit: `sleepAndGrow` both sleeps and returns the next backoff value, which
is a bit surprising from the name/signature alone (name suggests it just grows
the value). Might be worth splitting into a `sleep(long)` and a separate
`growBackoff(long)`, or renaming to make the side effect + return value clearer.
<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]
