Prevent duplicate data from zombie executor writes via intermediate file staging
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
### Feature Description
**What the feature achieves:**
Introduces an intermediate file staging mechanism that prevents zombie/stray Spark executors from leaving duplicate data files visible to query engines. When enabled, data files are written to a staging area (the marker directory) with a distinguishing prefix, and only promoted to the query-visible data partition during finalizeWrite. Files left behind by zombie executors are automatically cleaned up when the marker directory is deleted in postCommit.
**Why this feature is needed:**
In Hudi's Spark commit protocol, when a stage is retried, executors from the failed attempt (zombies) may continue running and finalize their data files after finalizeWrite has already completed -- or even after the commit is written. Because Hudi's file naming encodes taskAttemptId in the write token, these zombie-written files have unique names and are not overwritten by the successful attempt. They become visible to query engines, causing duplicate data.
This is particularly problematic on cloud object stores like GCS where:
There is no "deletes win" semantic (unlike HDFS NameNode)
Rename is not atomic without Hierarchical Namespace (HNS) enabled
Eventually-consistent listings can miss cleanup operations
The current reconciliation in finalizeWrite only deletes files not referenced in WriteStatus, but a zombie that closes its file after reconciliation bypasses this check entirely.
**Scope of impact:**
Affects all Hudi write operations (insert, upsert, bulk_insert) on Spark
Most severe on cloud object stores (GCS, S3) but also possible on HDFS under heavy contention
Can cause silent data duplication that is difficult to detect and repair
### User Experience
### User Experience
**How users will use this feature:**
**Configuration:** Enable via a single config property:
`hoodie.write.intermediate.file.enabled = true`
Default is false for safe, controlled rollout.
**No API changes:** Transparent to all existing write paths (insert, upsert, bulk_insert, compaction). WriteStatus always reports the final data partition path regardless of whether staging is enabled.
**No query-side changes:** Query engines (Spark, Presto, Trino, Hive) continue reading from data partitions as before. Intermediate files in the marker directory are invisible to queries.
**Cleanup is automatic:** Zombie intermediate files are cleaned up by the existing quietDeleteMarkerDir() call in postCommit. No additional cleanup configuration or operations are needed.
**Rollback safe:** If the process crashes before commit, the marker directory (containing intermediate files) is cleaned up by Hudi's existing rollback mechanism.
**Usage example:**
HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
.withPath(basePath)
.withIntermediateFileWriteEnabled(true)
.build();
### Implementation Summary
**Write phase:**
Data files are created in .hoodie/.temp/// with __hudi_intermediate_ prefix
The data partition directory is NOT created eagerly during write.
**Finalize phase (finalizeWrite):**
Files referenced in WriteStatus are renamed from the marker directory to the data partition
Target directory is created via mkdirs before rename
Renames are parallelized using HoodieEngineContext
**Cleanup:**
postCommit calls quietDeleteMarkerDir() which recursively deletes the marker directory, cleaning up any orphaned intermediate files from zombies
No additional sweep logic needed in clean or rollback
**Rename cost:**
HDFS: metadata-only operation (sub-millisecond)
GCS with HNS: metadata-only operation
GCS without HNS: copy + delete (bounded by file size, but acceptable for correctness)
### Hudi RFC Requirements
### RFC Requirements
RFC PR link: N/A
**Why RFC is/isn't needed:**
**Does this change public interfaces/APIs?** No -- adds a new opt-in config property only
**Does this change storage format?** No -- final data files have identical names and locations
**Justification:** This is a targeted correctness fix behind a feature flag. It reuses existing infrastructure (marker directory, finalizeWrite, postCommit cleanup) and does not alter the commit protocol, storage layout, or public APIs. An RFC is not required.
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by tracing HoodieWriteConfig and the Spark commit protocol through finalizeWrite, postCommit, and quietDeleteMarkerDir. Verify how the marker directory, __hudi_intermediate_ prefix, and WriteStatus paths are handled for staging and promotion. Done means the opt-in configuration stages files away from query-visible partitions, promotes referenced files during finalizeWrite, and cleans up orphaned zombie files during postCommit.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, spark
- Domain
- data-engineering, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100