hudi-agent commented on code in PR #20323:
URL: https://github.com/apache/hudi/pull/20323#discussion_r4267931750
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java:
##########
@@ -587,11 +595,10 @@ private void handleBootstrapEvent(WriteMetadataEvent
event) {
this.eventBuffers.cleanLegacyEvents(event);
return;
}
- EventBuffer eventBuffer =
this.eventBuffers.getOrCreateBootstrapBuffer(event);
- eventBuffer.addBootstrapEvent(event);
- if (eventBuffer.allBootstrapEventsReceived()) {
+ EventBuffer bootstrapBuffer = this.eventBuffers.addBootstrapEvent(event);
Review Comment:
🤖 On a partial failover of a non-RLI table, I think this can bring back a
buffer for an instant that is already committed: checkpoint 42 completes, and
`notifyCheckpointComplete(42)` starts committing buffer 41 on the executor.
While that runs, task 0 fails, restores from 42 (attempt 1), and still sees
instant 41 as pending, so it resends ckpt 41. The commit then removes buffer
41, and this event's `computeIfAbsent` recreates it holding only task 0's
statuses. On the next checkpoint, `commitInstants(43)` calls `doCommit` again
on the completed instant, and with a commit guard,
`awaitAllInstantsToCompleteIfNecessary` can block on it. Could you drop the
bootstrap event here when no buffer exists for its checkpoint and the reloaded
timeline already shows the instant as completed?
<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/utils/EventBuffers.java:
##########
@@ -91,11 +99,32 @@ public void addEventsToBuffer(Map<Long, Pair<String, ?>>
events) {
}
/**
- * Returns existing bootstrap buffer or creates a new one.
+ * Adds recovered metadata when its checkpoint is newer than the writer's
last accepted checkpoint.
+ * Returns null for duplicate or stale deliveries, including retries after
commit.
+ *
+ * <p>Each writer sends one merged bootstrap event per restored attempt.
Local retries restore the
+ * same or a newer completed checkpoint; global recovery resets this
tracking before replay.
+ * Data and index writers advance independently.
*/
- public EventBuffer getOrCreateBootstrapBuffer(WriteMetadataEvent event) {
- return this.eventBuffers.computeIfAbsent(event.getCheckpointId(),
+ @Nullable
+ public EventBuffer addBootstrapEvent(WriteMetadataEvent event) {
+ long[] checkpoints = event.isMetadataTable() ? indexBootstrapCheckpoints :
dataBootstrapCheckpoints;
+ if (event.getCheckpointId() <= checkpoints[event.getTaskID()]) {
+ return null;
+ }
+ EventBuffer buffer =
this.eventBuffers.computeIfAbsent(event.getCheckpointId(),
Review Comment:
🤖 nit: could you name the `checkpoints` local `writerCheckpoints` (or
similar)? It is a per-writer array indexed by task ID, and the generic name
hides that.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/common/TestAbstractStreamWriteFunction.java:
##########
@@ -133,9 +133,9 @@ void testRestoredMetadataKeepsOriginalCheckpointId() throws
Exception {
writeMetadataState().add(restored);
when(pendingTimeline.containsInstant("001")).thenReturn(true);
- initialize(42L, 0);
+ initialize(42L, attempt);
- assertEquals(1, events.size());
+ assertEquals(attempt == 0 ? 1 : 2, events.size());
Review Comment:
🤖 nit: `attempt == 0 ? 1 : 2` is a bit cryptic; would it make sense to add a
short comment, or a named `expectedEvents` variable, saying the extra event is
the empty cleanup bootstrap on retries?
<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]
