fix(flink): reuse coordinator pending fileIds to avoid duplicate bucket fileId after recommit race - #19951
fix(flink): reuse coordinator pending fileIds to avoid duplicate bucket fileId after recommit race#19951zhaoyudi-creator wants to merge 2 commits into
Conversation
…et fileId after recommit race With the simple Bucket Index, a write task's bootstrapIndexIfNeed() loads bucket -> fileId only from the committed file-system view. After a failover before the coordinator commits the first write to a partition, the fileId flushed to the still-inflight instant is invisible to the restarted task, so it mints a brand-new fileId for the same bucket. Once the coordinator recommits the restored bootstrap event, the bucket ends up owning two fileIds and the next bootstrap fails with "Duplicate fileId ... found". Have bootstrap additionally query the coordinator for the fileIds it still holds pending (checkpointed but not yet committed) and overlay them onto buckets this task owns that the committed view left empty. The committed view wins on conflict, so only otherwise-empty buckets adopt a pending fileId. The query runs on the coordinator's meta-event executor, serialized behind any enqueued bootstrap-event handling, so a pending fileId is observed even while its recommit is still in flight. The path does not touch async instant generation: the query is a read-only lookup issued once per partition on first bootstrap, off the per-record hot path, and adds no blocking to the commit pipeline. Closes apache#19907
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR has the simple Bucket Index bootstrap ask the coordinator for fileIds it still holds in uncommitted event buffers, so a restarted task adopts the pending fileId instead of minting a second one for the same bucket. The main thing worth double-checking in the inline comments is the ordering between the committed-view load and the coordinator query relative to the recommit — in the real runtime both the bootstrap event and this new request go through the same single-threaded coordinator executor, which I think leaves the original window open for the task that triggers the recommit (the new test only passes because it holds the bootstrap event back). There's also a question on the per-partition blocking round-trip cost. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of minor nits around returning/storing the concrete HashSet type instead of the Set interface in the new pending-fileIds plumbing; otherwise the change is clean and well-commented.
| } | ||
| } | ||
| }); | ||
| overlayPendingBucketFileIds(partition, bucketToFileIDMap); |
There was a problem hiding this 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 passes 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.
| if (this.correspondent == null) { | ||
| return; // not wired in some tests; nothing to overlay | ||
| } | ||
| Set<String> pendingFileIds = this.correspondent.requestPendingBucketFileIds(partition); |
There was a problem hiding this 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?
| .flatMap(eventBuffer -> Arrays.stream(eventBuffer.getDataWriteEventBuffer())) | ||
| .filter(Objects::nonNull) | ||
| .flatMap(event -> event.getWriteStatuses().stream()) | ||
| .filter(writeStatus -> partition.equals(writeStatus.getPartitionPath())) |
There was a problem hiding this comment.
🤖 nit: the method exposes HashSet<String> in its public signature — consider returning Set<String> instead, consistent with Correspondent.requestPendingBucketFileIds which already returns Set.
|
|
||
| private final HashSet<String> fileIds; | ||
|
|
||
| public static PendingBucketFileIdsResponse getInstance(HashSet<String> fileIds) { |
There was a problem hiding this 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).
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19951 +/- ##
============================================
- Coverage 80.32% 80.31% -0.01%
- Complexity 34744 34759 +15
============================================
Files 2546 2546
Lines 142515 142559 +44
Branches 17303 17317 +14
============================================
+ Hits 114476 114502 +26
- Misses 20136 20155 +19
+ Partials 7903 7902 -1
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
| return CompletableFuture.completedFuture(CoordinationResponseSerDe.wrap(coordinationResponse)); | ||
| } | ||
|
|
||
| private CompletableFuture<CoordinationResponse> handlePendingBucketFileIdsRequest(Correspondent.PendingBucketFileIdsRequest request) { |
There was a problem hiding this comment.
The reported issue is valid: partition bootstrap can run before the coordinator finishes recommitting the first write, causing the writer to generate a second fileId for the same bucket.
However, this implementation still leaves that race open: the writer can read the committed view before recommit finishes, then execute this lookup after recommit has completed and cleared the event buffer. Both reads miss the original fileId, causing the writer to generate another fileId for the same bucket. Serializing this lookup with recommit on the same executor does not prevent that sequence.
One possible solution is: we maintain a separate partition-> bucketId -> fileId recovery mapping, populated from restored coordinator state and incoming bootstrap events, and retain it for the coordinator’s lifetime? Bootstrap could merge this mapping with the committed view, giving committed entries precedence. This would preserve the original fileId across pending-buffer cleanup without waiting for recommit.
There was a problem hiding this comment.
Thanks @cshuo. Your analysis is correct for the original version — with the committed view loaded before the pending-fileId lookup, it does produce a duplicate fileId.
The latest commit reverses the order in bootstrapIndexIfNeed: query pending fileIds first, then load the committed view from a fresh getHoodieTable(). Since the lookup and the recommit run on the same single-threaded executor (FIFO), and commit lands before the buffer reset in doCommit, both orderings recover fileId-A: lookup before recommit → the buffer still holds it; lookup after recommit → it returns empty, but "empty" already implies the commit landed, so the committed view loaded afterwards is guaranteed to see it. The "both miss" case can no longer be constructed.
That said, I don't want to just push my own direction — I'd like to weigh the two approaches with you:
The reorder in this PR: lightweight, adds no coordinator state, but correctness leans on two implicit assumptions (lookup and recommit on the same executor; getHoodieTable() not caching a stale timeline) that a future refactor could quietly break.
Your coordinator-lifetime partition → bucketId → fileId mapping: relies on no timing reasoning and is more robust, at the cost of extra coordinator state to maintain correctly.
Which do you think fits better here? Would love your read, based on the latest commit.
There was a problem hiding this comment.
@zhaoyudi-creator Thanks for the update. Querying pending fileIds before loading a fresh committed view addresses the original missing-fileId window, assuming the relevant recovery metadata is already registered. I still have two concerns:
- Blocking the writer’s write flow. The lookup shares the single-threaded executor with recommit. If recommit is running, the request waits behind it, and the writer synchronously waiting for the response cannot continue processing records. This effectively couples partition bootstrap to commit latency.
- RPC timeouts during slow recommits. This request uses Flink’s coordination RPC, whose ask timeout defaults to 10 seconds. A slow recommit can exhaust that timeout before the lookup executes, causing a task failure rather than just delaying bootstrap. #19902 reports a similar failure mode for instant-time requests waiting on coordinator-side work.
cc @danny0405
There was a problem hiding this comment.
Agreed on both concerns. The RPC timeout in particular is something we actually ran into during our 1.1 testing, so it's a real failure mode rather than a theoretical one. Let me rethink the approach.
There was a problem hiding this comment.
@danny0405 @cshuo Thanks for the careful review — both concerns are valid, and following them through leads to a conclusion worth stating explicitly.
On restore the coordinator recommits in two places:
Path 1 — restoreEvents: executed synchronously inside start() / resetToCheckpoint(). Subtasks are deployed only after it returns successfully, which means Path 1's recommit always completes before bootstrapIndexIfNeed runs.
Path 2 — recommitInstant inside handleBootstrapEvent: dispatched via executor.execute, where buffer population and recommit run in the same task and share the single-threaded executor with the lookup. This is exactly where it can go wrong: a slow recommit blocks the lookup and can exhaust the 10s ask timeout.
Path 2 also exposes a correctness window under rescale. Under rescale the mapping changes, so the task that owns the bucket and issues the lookup is generally not the task that resends fileId-A. Two different subtasks have no ordering guarantee toward the coordinator.
Conclusion: a lookup that does not sit on the commit path cannot guarantee the absence of duplicate fileIds — blocking and correctness are two ends of the same trade-off. That leaves only two ways forward:
- Wait for the recommit to succeed — have the lookup explicitly wait for the instant to be recommitted (accepting the coupling with commit latency);
- Deterministic fileId — per-bucket, though the insert-overwrite scenario needs to be considered.
…estore Address review feedback on the bucket-index recommit-race fix: - Query the coordinator for pending fileIds before loading the committed view so an empty result guarantees the freshly reloaded view observes any in-flight recommit (happens-before via the coordinator executor). - Gate the pending-fileId query on isRestored so a fresh start pays no extra RPC and the async instant-generation path stays non-blocking. - Widen the pending-fileId API types from HashSet to Set. - Rewrite the tests to exercise both real orderings: task-level failover (pending query is the sole protection) and full job restart (committed view fallback after a genuine recommit). Closes apache#19907
github-actions
Bot
added
size:L
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for addressing the feedback! The update reorders the bootstrap so the pending-fileId query hits the coordinator before the committed view is built, gates the query on context.isRestored(), widens the HashSet signatures to Set, and adds a second test that delivers the bootstrap event to the coordinator before the record (the runtime ordering).
On the prior comments: the ordering concern (mine and @cshuo's) is addressed — I traced it again and with FIFO on the coordinator executor the two cases are now consistent: a query that returns empty implies the recommit already ran, and writeClient.getHoodieTable() builds a fresh metaClient/timeline so the view sees it; a query that returns the pending id adopts it directly. I also checked the task-level failover case in the first test: the ckp-1 flush event is buffered under the previous checkpoint id, so the empty bootstrap's cleanLegacyEvents (which clears keys >= restored id) leaves it pending, meaning that test's scenario is faithful even without holding the bootstrap event back. The per-partition RPC question is addressed by the isRestored gate, and both Set<String> nits are done. No new issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
|
|
||
| // Load existing fileID belongs to this task | ||
| Map<Integer, String> bucketToFileIDMap = new HashMap<>(); | ||
| this.writeClient.getHoodieTable().getHoodieView().getLatestFileSlices(partition).forEach(fileSlice -> { |
There was a problem hiding this comment.
can we just change it to getLatestFileSlicesIncludingInflight to fix the issue?
There was a problem hiding this comment.
getLatestFileSlicesIncludingInflight should still miss the inflight fileId when the metadata table is enabled. With metadata.enabled=true (the Flink default), the view's file groups are built from the MDT FILES-partition listing (FileSystemViewManager#getTableMetadata → AbstractTableFileSystemView#ensurePartitionLoadedCorrectly → BaseTableMetadata#fetchAllFilesInPartition), which only reflects committed files. Since the inflight instant has no MDT record yet, its file isn't in any file group, and getLatestFileSlicesIncludingInflight — which only drops the committed filter — can't surface it.
This branch has not been deployed
6 participants
Describe the issue this Pull Request addresses
With Flink streaming writes and the simple Bucket Index (non-NBCC), if a failover/restore
happens while the first write to a partition is still uncommitted, the restarted task reads
only the committed file-system view (still empty at that point) and mints a brand-new fileId for
the bucket. This collides with the inflight fileId the coordinator later recommits, and the next
bootstrap fails with
Duplicate fileId ... found. See the issue for the full reproduction timeline.Summary and Changelog
Fixes a duplicate-fileId failure that can permanently break bootstrap of a simple Bucket Index
partition after a failover. The fix has
bootstrapIndexIfNeed(), in addition to reading thecommitted view, query the coordinator for the fileIds it still holds pending (checkpointed but not
yet committed) and reuse them, so a restarted task adopts the original fileId instead of minting a
new one.
I'd appreciate a maintainer's read on the direction before I invest further. An alternative would
be a deterministic fileId (like NBCC's
-0000-...), which is a smaller change but changes theon-disk fileId format, diverges from the non-NBCC behavior of Spark/bulk-insert, and cannot recover
an already-existing inflight bucket that used a random id across an upgrade. The coordinator-query
approach here keeps the format unchanged and naturally covers rescale (after a bucket's owner
changes, the new owner adopts the pending fileId). If this direction is acceptable, I'll extend the
test coverage (see below).
Planned follow-up tests if the approach is accepted:
(committed view wins, overlay does not overwrite an existing fileId).
repeat across partitions (the query filters by partition and does not cross fileIds).
better suited as an IT).
Impact
No public API or user-facing change. Adds one internal coordination request that is read-only and
issued only once per partition on first bootstrap (guarded by
bucketIndex.containsKey(partition)).It is off the per-record hot path and adds no blocking to the commit pipeline, so it does not affect
the async instant-generation optimization. NBCC and other index types are unaffected.
Risk Level
Documentation Update
Contributor's checklist