hudi-agent commented on code in PR #19951:
URL: https://github.com/apache/hudi/pull/19951#discussion_r4012126706
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bucket/BucketStreamWriteFunction.java:
##########
@@ -206,6 +206,26 @@ private void bootstrapIndexIfNeed(String partition) {
}
}
});
+ overlayPendingBucketFileIds(partition, bucketToFileIDMap);
Review Comment:
π€ I think the order here still leaves the original window open in the real
runtime. The bootstrap event is sent from `initializeState` before any record
is processed, and both `handleEventFromOperator` and the new
`handlePendingBucketFileIdsRequest` are queued on the same single-threaded
`executor` β so for the task whose bootstrap event completes the set (always,
with parallelism 1), the coordinator runs `recommitInstant` β `doCommit` β
`eventBuffers.reset` *before* it answers this request, and the request comes
back empty. Meanwhile the committed view above was loaded before the query
(possibly before the commit landed), so it's empty too and we still mint a
fresh fileId. Could you query the pending fileIds *first* and then build the
committed view from the fresh `getHoodieTable()`? With FIFO on the executor
that gives a consistent answer either way: if the recommit already ran, the
fresh view sees it; if not, the query still returns the pending id. Note the
test only p
asses because step 2 holds the bootstrap event back from the coordinator,
which doesn't match runtime ordering. @danny0405 would appreciate your read on
the direction 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/bucket/BucketStreamWriteFunction.java:
##########
@@ -206,6 +206,26 @@ private void bootstrapIndexIfNeed(String partition) {
}
}
});
+ overlayPendingBucketFileIds(partition, bucketToFileIDMap);
bucketIndex.put(partition, bucketToFileIDMap);
}
+
+ /**
+ * Overlay fileIds the coordinator still holds pending (flushed to an
inflight instant but not yet
+ * committed): the committed view above cannot see them, so a restarted task
would mint a second
+ * fileId for a bucket that already owns a pending one, corrupting it on the
recommit.
+ */
+ private void overlayPendingBucketFileIds(String partition, Map<Integer,
String> bucketToFileIDMap) {
+ if (this.correspondent == null) {
+ return; // not wired in some tests; nothing to overlay
+ }
+ Set<String> pendingFileIds =
this.correspondent.requestPendingBucketFileIds(partition);
Review Comment:
π€ This runs on every `bootstrapIndexIfNeed`, i.e. a blocking round-trip to
the coordinator for every partition a subtask sees for the first time β not
just after a restore. Since the request is answered on the coordinator's
single-threaded `executor`, the first record of each partition can stall behind
an in-progress commit (seconds with MDT). Have you considered only issuing it
when the task restored from state (or once per task, fetching partitionβfileIds
for all partitions), so steady-state jobs with many partitions don't pay it?
<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:
##########
@@ -153,4 +170,32 @@ public static InflightInstantsResponse
getInstance(HashMap<Long, String> infligh
return new InflightInstantsResponse(inflightInstants);
}
}
+
+ /**
+ * A request for the fileIds pending (checkpointed but not yet committed)
for a partition.
+ */
+ @AllArgsConstructor(access = AccessLevel.PRIVATE)
+ @Getter
+ public static class PendingBucketFileIdsRequest implements
CoordinationRequest {
+
+ private final String partition;
+
+ public static PendingBucketFileIdsRequest getInstance(String partition) {
+ return new PendingBucketFileIdsRequest(partition);
+ }
+ }
+
+ /**
+ * A response with the fileIds pending for a partition.
+ */
+ @AllArgsConstructor(access = AccessLevel.PRIVATE)
+ @Getter
+ public static class PendingBucketFileIdsResponse implements
CoordinationResponse {
+
+ private final HashSet<String> fileIds;
+
+ public static PendingBucketFileIdsResponse getInstance(HashSet<String>
fileIds) {
Review Comment:
π€ nit: `PendingBucketFileIdsResponse` stores/accepts a `HashSet<String>` β
worth using `Set<String>` for the field and `getInstance` param to match the
`Set<String>` return type used elsewhere in this file (e.g.
`requestPendingBucketFileIds`).
<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:
##########
@@ -175,6 +176,22 @@ public Map<Long, Pair<String, EventBuffer>>
getAllCompletedEvents() {
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
+ /**
+ * Collects the fileIds carried by all pending (buffered but not yet
committed) data write events for
+ * {@code partition}.
+ */
+ public HashSet<String> getPendingWriteFileIds(String partition) {
+ HashSet<String> fileIds = new HashSet<>();
+ this.eventBuffers.values().stream()
+ .map(Pair::getRight)
+ .flatMap(eventBuffer ->
Arrays.stream(eventBuffer.getDataWriteEventBuffer()))
+ .filter(Objects::nonNull)
+ .flatMap(event -> event.getWriteStatuses().stream())
+ .filter(writeStatus ->
partition.equals(writeStatus.getPartitionPath()))
Review Comment:
π€ nit: the method exposes `HashSet<String>` in its public signature β
consider returning `Set<String>` instead, consistent with
`Correspondent.requestPendingBucketFileIds` which already returns `Set`.
<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]
