hudi-agent commented on code in PR #19573: URL: https://github.com/apache/hudi/pull/19573#discussion_r3753941406
########## website/docs/hoodie_streaming_ingestion.md: ########## @@ -312,14 +312,54 @@ Read more in depth about concurrency control in the [concurrency control concept Hudi Streamer uses checkpoints to keep track of what data has been read already so it can resume without needing to reprocess all data. When using a Kafka source, the checkpoint is the [Kafka Offset](https://cwiki.apache.org/confluence/display/KAFKA/Offset+Management) When using a DFS source, the checkpoint is the 'last modified' timestamp of the latest file read. -Checkpoints are saved in the .hoodie commit file as `streamer.checkpoint.key`. +Checkpoints are saved in the .hoodie commit file, under `streamer.checkpoint.key.v2` for tables at table +version 8 or higher and `deltastreamer.checkpoint.key` below that. Review Comment: 🤖 This line reads as though every source's checkpoint moves to `streamer.checkpoint.key.v2` at table version 8+, but that isn't what the code does — and it sits directly after the Kafka and DFS descriptions, which is exactly where a reader will apply it. In `StreamerCheckpointUtils.shouldTargetCheckpointV2()` the V2 key is used only for the `HoodieIncrSource` family (and excludes `S3EventsHoodieIncrSource`/`GcsEventsHoodieIncrSource`); `StreamSync.getCheckpointCommitMetadata()` otherwise falls back to `createCheckpoint()`, which always produces a `StreamerCheckpointV1` writing under `deltastreamer.checkpoint.key`. So a Kafka or DFS checkpoint stays under `deltastreamer.checkpoint.key` even on a table-version-8+ table. It would help to scope this sentence to the Hudi incremental source rather than to table version alone, so Kafka/DFS users don't go looking for a `.v2` key that never gets written. @yihua could you confirm the intended key semantics here — the checkpoint class es carry your TODOs, so you'd know best whether V2 keys are ever written for non-incremental sources at v8+? (Same wording appears in the version-1.0.1/1.0.2/1.1.1/1.2.0 copies.) <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## website/docs/hoodie_streaming_ingestion.md: ########## @@ -312,14 +312,54 @@ Read more in depth about concurrency control in the [concurrency control concept Hudi Streamer uses checkpoints to keep track of what data has been read already so it can resume without needing to reprocess all data. When using a Kafka source, the checkpoint is the [Kafka Offset](https://cwiki.apache.org/confluence/display/KAFKA/Offset+Management) When using a DFS source, the checkpoint is the 'last modified' timestamp of the latest file read. -Checkpoints are saved in the .hoodie commit file as `streamer.checkpoint.key`. +Checkpoints are saved in the .hoodie commit file, under `streamer.checkpoint.key.v2` for tables at table +version 8 or higher and `deltastreamer.checkpoint.key` below that. If you need to change the checkpoints for reprocessing or replaying data you can use the following options: -- `--checkpoint` will set `streamer.checkpoint.reset_key` in the commit file to overwrite the current checkpoint. Format of checkpoint depends on [KAFKA_CHECKPOINT_TYPE](configurations.md#hoodiestreamersourcekafkacheckpointtype). By default (for type `string`), checkpoint should be provided as: `topicName,0:offset0,1:offset1,2:offset2`. For type `timestamp`, checkpoint should be provided as long value of desired timestamp. For type `single_offset`, we assume that topic consists of a single partition, so checkpoint should be provided as long value of desired offset. +- `--checkpoint` will set the checkpoint reset key in the commit file to overwrite the current checkpoint — `streamer.checkpoint.reset.key.v2` at table version 8 or higher, `deltastreamer.checkpoint.reset_key` below that. Format of checkpoint depends on [KAFKA_CHECKPOINT_TYPE](configurations.md#hoodiestreamersourcekafkacheckpointtype). By default (for type `string`), checkpoint should be provided as: `topicName,0:offset0,1:offset1,2:offset2`. For type `timestamp`, checkpoint should be provided as long value of desired timestamp. For type `single_offset`, we assume that topic consists of a single partition, so checkpoint should be provided as long value of desired offset. - `--source-limit` will set a maximum amount of data to read from the source. For DFS sources, this is max # of bytes read. For Kafka, this is the max # of events to read. Review Comment: 🤖 Same root cause as the intro line: this bullet now says `--checkpoint` sets `streamer.checkpoint.reset.key.v2` at table version 8+, but `buildCheckpointFromConfigOverride()` only returns the V2-capable (`UnresolvedStreamerCheckpointBasedOnCfg`) checkpoint for the `HoodieIncrSource` family — for Kafka/DFS sources it returns `StreamerCheckpointV1`, whose `getCheckpointCommitMetadata()` writes the reset value under `deltastreamer.checkpoint.reset_key` regardless of table version. Since this bullet is describing the Kafka-format `--checkpoint` value, the reset key it will actually see is `deltastreamer.checkpoint.reset_key`, not the `.v2` one. Consider limiting the `.v2` reset-key mention to the Hudi incremental source (where the new subsection already covers it) and keeping the V1 key for the Kafka/DFS bullet. This is a different point from the earlier name-reconciliation thread — the concern here is that the `.v2` name is wrong for the non-incremental sources this bullet do cuments. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- 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]
