hudi-agent commented on code in PR #19960:
URL: https://github.com/apache/hudi/pull/19960#discussion_r4078336453
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java:
##########
@@ -412,22 +425,47 @@ 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.
+ final 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);
+ }
+ return CompletableFuture.completedFuture(
Review Comment:
🤖 nit: since there's a `readyResponse(...)` helper, could you add a matching
`pendingResponse()` so both branches read symmetrically 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/common/AbstractStreamWriteFunction.java:
##########
@@ -266,7 +267,8 @@ public void handleOperatorEvent(OperatorEvent event) {
* @return The instant time
*/
protected String instantToWrite(boolean hasData) {
- return
Preconditions.checkNotNull(this.correspondent.requestInstantTime(this.checkpointId),
+ return Preconditions.checkNotNull(
+ this.correspondent.requestInstantTime(this.checkpointId,
this.config.get(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT)),
Review Comment:
🤖 nit: `WRITE_COMMIT_ACK_TIMEOUT` is now also the instant-request poll
budget; it might be worth a short comment here (and in BulkInsertWriteFunction)
explaining the reuse so readers don't assume it's only about commit acks.
<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]
