zhaoyudi-creator opened a new issue, #19907: URL: https://github.com/apache/hudi/issues/19907
### Describe the problem you faced With **Bucket Index + Flink streaming write**, when a partition is written for the **first time** (the bucket has no committed base file yet), if the write task undergoes failover / savepoint restore before its corresponding instant is committed, a later index bootstrap throws: ``` Duplicate fileId <fileId> from bucket <bucketNo> of partition <partition> found during the BucketStreamWriteFunction index bootstrap. ``` ### To Reproduce Verified online timeline: 1. **16:37** — Job restarts and recovers from state; TM re-sends the bootstrap event for the inflight instant `20260605161125980`. 2. **~16:37** — Write side: first record of the new partition → `bootstrapIndexIfNeed`; `20260605161125980` is still inflight → committed view is empty → allocates new **fileId-B**. 3. **16:40:14** — JM commits `20260605161125980` via the **recommit route** (`restoreEvents` / bootstrap quorum → `recommitInstant` → `commitInstant`), logs `Recommit instant 20260605161125980`; **fileId-A** lands. 4. **16:40:18** — JM opens the new instant `20260605164017942` (`startInstant`). 5. **16:40:19** — A second recommit trigger enters `recommitInstant`; since `20260605161125980` is already completed it falls into the else-branch `reset` (harmless cleanup, confirming the commit went through the recommit route). 6. **Later** — fileId-B is committed under `20260605164017942`; the bucket now holds fileId-A + fileId-B, and the next bootstrap throws `Duplicate fileId`. ### Expected behavior After failover / savepoint restore, the write-side index bootstrap should see the bucket's still-pending fileId and reuse the same fileId, rather than creating a second file group. ### Environment Description * Hudi version: * Spark version: * Flink version: * Hive version: * Hadoop version: * Storage (HDFS/S3/GCS..): * Running on Docker? (yes/no): ### Additional context ## Root cause — two execution lines that do not wait for each other: - **Write-task side:** after restore, `bootstrapIndexIfNeed()` loads the bucket→fileId map from `getLatestFileSlices()` (**committed view only**). The old instant is still inflight, so fileId-A is invisible; `defineRecordLocation()` then allocates a **new fileId-B** (random UUID suffix) for the same bucket. - **Coordinator side:** it commits that old inflight instant **through the recommit route** (`recommitInstant`), **asynchronously and after** the write side has already allocated fileId-B. fileId-B is later committed under the new instant. - **Result:** the bucket ends up with two file groups (fileId-A + fileId-B), and the next bootstrap hits the `Duplicate fileId` check. ## Proposed fix & open questions **Proposed fix:** In `bootstrapIndexIfNeed`, extend the source of the bucket→fileId map from "committed view only" to "committed ∪ fileIds written by still-pending (inflight) instants". This way the still-inflight fileId-A of a first-written partition becomes visible to bootstrap and is reused, instead of minting a new fileId-B. The approach is **non-blocking** (it does not intercept the consume path) and holds for both the `attemptId ≤ 0` and `attemptId > 0` restore branches. **Downside:** It needs an API with proper **filtering semantics** to fetch the file slices of pending instants — the unfiltered `getLatestFileSlicesIncludingInflight` cannot be used directly, otherwise orphan files from instants that are about to be rolled back would be pulled in as valid fileIds. If no such clean API exists, the fallback is to have the coordinator broadcast the pending fileIds it already holds to the write tasks for warm-up, which is a larger change. **We'd like the community to confirm:** 1. Is this race **actually real** (rather than a false problem already covered by some other mechanism)? 2. Is there a **better approach** (e.g. an existing filtering API, or a more suitable layer to fix it at)? ### Stacktrace ```shell ``` -- 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]
