apache / apache/hudi

HoodieStreamer persists V2 checkpoint key on table-version-6 writes

Open
#18,888 0 comments 0 reactions 0 assignees View on GitHub
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.