HoodieStreamer persists V2 checkpoint key on table-version-6 writes
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
### Tips before filing an issue
- [x] Have you gone through our [FAQs](https://hudi.apache.org/learn/faq/)?
### Describe the problem you faced
When `HoodieStreamer` ingests with `hoodie.write.table.version=6` and the source is one that funnels a `StreamerCheckpointV2` through `InputBatch#getCheckpointForNextBatch` (e.g. any `*DFSSource` via `DFSPathSelector`, Kafka via the `InputBatch(batch, String)` constructor, Kinesis, Pulsar, Jdbc, Sql, Hive, HoodieIncrSource, DatePartitionPathSelector, Debezium, S3EventsMetaSelector, GcsEventsSource), the persisted commit metadata carries the V2 key `streamer.checkpoint.key.v2` instead of the V1 key `deltastreamer.checkpoint.key`.
This is a contract violation: `CheckpointUtils.shouldTargetCheckpointV2(writeTableVersion, sourceClassName)` returns `false` for `writeTableVersion < 8`, so V1 is required for v6 tables.
### To Reproduce
Steps to reproduce the behavior:
1. Use HoodieStreamer with `--source-class org.apache.hudi.utilities.sources.ParquetDFSSource`
2. Pass `--hoodie-conf hoodie.write.table.version=6`
3. After a sync produces a commit, inspect `extraMetadata` in the `.commit` file
4. Observe `streamer.checkpoint.key.v2=` is present and `deltastreamer.checkpoint.key` is absent
### Expected behavior
For `hoodie.write.table.version=6`, the persisted commit metadata should carry the V1 key (`deltastreamer.checkpoint.key`) and not the V2 key (`streamer.checkpoint.key.v2`).
### Environment Description
- Hudi version: master (1.x line)
- Spark version: 3.5
- Hive version: n/a
- Hadoop version: n/a
- Storage (HDFS/S3/GCS..): any
- Running on Docker?: no
### Additional context
Root cause: `DFSPathSelector.getNextFilePathsAndMaxModificationTime` (lines 154 & 160) and similar helpers in other sources unconditionally construct `new StreamerCheckpointV2(...)` without checking the target table version. `StreamSync.extractCheckpointMetadata` then trusts that type and writes V2 keys regardless of write table version. The branch two lines below for the no-batch fallback already calls `buildCheckpointFromGeneralSource` and gets the version contract right — so this is a missed branch in an existing chokepoint, not a missing primitive.
PR: WIP — will link.
### Stacktrace
n/a — wrong-key persistence, not a thrown exception.
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with DFSPathSelector.getNextFilePathsAndMaxModificationTime and compare its checkpoint construction with the nearby no-batch fallback that calls buildCheckpointFromGeneralSource. Trace how StreamSync.extractCheckpointMetadata persists the checkpoint, reproduce with HoodieStreamer, ParquetDFSSource, and table version 6, then verify the commit metadata contains deltastreamer.checkpoint.key instead of streamer.checkpoint.key.v2.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, spark
- Domain
- stream-processing
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 52/100