hudi-agent commented on code in PR #19960:
URL: https://github.com/apache/hudi/pull/19960#discussion_r4072338741
##########
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:
🤖 Dropping the post-submit re-lookup (from the earlier suggestion) means the
submitting request always gets `PENDING`, so the writer's first chance at
`READY` is only after `sleepAndGrow`. With the synchronous
`MockCoordinatorExecutor`, `createInstant` (including `startInstant()`'s
filesystem work) runs inside that first call, so a 1ms
`write.commit.ack.timeout` is already past the deadline when `PENDING` comes
back — I ran `TestWriteCopyOnWrite#testCheckpointFails` and `#testSubtaskFails`
on this revision and all three error with `Timeout waiting for the instant time
from the coordinator for checkpoint -1` (they were green on the previous push).
In production it also adds a mandatory ≥25–50ms sleep to every fresh instant
request. Could we restore the cheap `getInstantAndEventBuffer` check after
`execute(...)` and return `READY` when it's already published, or alternatively
make the writer loop guarantee at least one re-poll before enforcing the
deadline?
<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: reusing `WRITE_COMMIT_ACK_TIMEOUT` as the instant-poll budget isn't
obvious from the option name — could you add a short comment here (and in
`BulkInsertWriteFunction`) explaining why the commit-ack timeout is the right
bound, or pull it into a small `instantRequestPollBudgetMs()` helper so the
intent lives in one place?
<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]
