cshuo commented on code in PR #19960:
URL: https://github.com/apache/hudi/pull/19960#discussion_r4069864873
##########
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());
+ }
Review Comment:
Could we remove this second lookup and return `PENDING` directly? The test
helper could use `MockCorrespondent.requestInstantTime()` to follow the polling
protocol. This simplifies the handler at the cost of one extra poll if creation
finishes immediately.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/EventBuffers.java:
##########
@@ -71,7 +72,13 @@ public static EventBuffers getInstance(Configuration conf,
int dataWriteParallel
}
public EventBuffer addEventToBuffer(WriteMetadataEvent event) {
- EventBuffer eventBuffer =
this.eventBuffers.get(event.getCheckpointId()).getRight();
+ Pair<String, EventBuffer> bufferPair =
this.eventBuffers.get(event.getCheckpointId());
+ ValidationUtils.checkState(bufferPair != null,
+ "No event buffer bound to checkpoint " + event.getCheckpointId());
+
ValidationUtils.checkState(bufferPair.getLeft().equals(event.getInstantTime()),
+ String.format("Event instant %s does not match the instant %s bound to
checkpoint %d",
+ event.getInstantTime(), bufferPair.getLeft(),
event.getCheckpointId()));
Review Comment:
Are these defensive validations necessary for this pr? If the new polling
flow requires them, could you clarify the scenario that can produce a missing
buffer or mismatched instant?
##########
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;
Review Comment:
Could we retry only `PENDING` responses for now and fail fast on exceptions?
The broad `catch (Exception)` also retries cancellation and deserialization
failures, which are unlikely to recover through retries. If we want to retry
transient RPC failures, we should explicitly classify retryable causes after
unwrapping `ExecutionException`.
##########
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:
This cap appears redundant: `backoffMs` starts below `POLL_CAP_MS`, and
`sleepAndGrow()` already caps its return value. Could we use `backoffMs`
directly when calculating the jitter?
--
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]
