deepakpanda93 commented on code in PR #19573: URL: https://github.com/apache/hudi/pull/19573#discussion_r3755257653
########## 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: You're right that these two lines were wrong, and thanks for pushing on it — but the measurement lands somewhere neither of us had it. Fixed in cc3483885c7c. **Where the review's mechanism doesn't match `release-1.2.0`.** There is no `createCheckpoint()` in `StreamSync` at this tag. The write path is `StreamSync.extractCheckpointMetadata` (L911-927): ```java if (inputBatch.getCheckpointForNextBatch() != null) { return inputBatch.getCheckpointForNextBatch().getCheckpointCommitMetadata(cfg.checkpoint, cfg.ignoreCheckpoint); } Checkpoint checkpoint = buildCheckpointFromGeneralSource(cfg.sourceClassName, versionCode, null); return checkpoint.getCheckpointCommitMetadata(cfg.checkpoint, cfg.ignoreCheckpoint); ``` The `buildCheckpointFromGeneralSource` branch — the one carrying the source-class exclusion — is only the *fallback*, used when the source supplied no checkpoint. In the normal case the key comes from whatever `Checkpoint` object the source returned. **What actually decides it.** `InputBatch(Option<T> batch, String checkpointForNextBatch)` delegates to `this(batch, new StreamerCheckpointV2(checkpointForNextBatch), ...)` (`InputBatch.java:43-44`) — wrapping in **V2 unconditionally**. Sources that hand back a plain string, Kafka and DFS among them, therefore always produce a V2 checkpoint. So Kafka/DFS do **not** stay on `deltastreamer.checkpoint.key`. I checked rather than reason about it, running `ParquetDFSSource` twice over the same input: | Target table version | Checkpoint key written | |---|---| | 9 (default) | `streamer.checkpoint.key.v2` | | 6 (`hoodie.write.table.version=6`) | `streamer.checkpoint.key.v2` | It writes the v2 key at **both**, which also makes my "version 8 or higher" rule wrong — in the opposite direction from the one suggested. `HoodieIncrSource` behaves differently because it returns a real `Checkpoint` object whose type *is* version-gated, which is why my earlier version-6 run of that source produced `deltastreamer.checkpoint.key`. **The fix.** The split is by source, and only *within* the Hudi incremental source by table version. Both lines now name the two keys and say what the choice depends on, without asserting a version rule that does not hold for Kafka or DFS: > Checkpoints are saved in the .hoodie commit file. Completion time checkpoints are stored under `streamer.checkpoint.key.v2` and request time checkpoints under `deltastreamer.checkpoint.key`. Which one applies depends on the source, and for the Hudi incremental source also on the table version — see below. > `--checkpoint` will set the matching reset key in the commit file to overwrite the current checkpoint, either `streamer.checkpoint.reset.key.v2` or `deltastreamer.checkpoint.reset_key`. The version-specific detail stays in the incremental source subsection, which is the part backed by the six-release matrix. Note the original text named `streamer.checkpoint.key` and `streamer.checkpoint.reset_key`, neither of which Hudi writes, so reverting was not an option. Build clean, warnings byte-identical to baseline. @yihua a committer's eye on the Kafka/DFS key behaviour would still be welcome — the `InputBatch` string constructor forcing V2 regardless of table version looks deliberate but is worth confirming. ########## 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, same fix — see the reply on the intro line for the detail and the measurements. Corrected in cc3483885c7c. Short version: `buildCheckpointFromConfigOverride` is not what decides this in the common path. `StreamSync.extractCheckpointMetadata` uses the `Checkpoint` object the source returned, and `InputBatch(batch, String)` wraps a plain string in `StreamerCheckpointV2` unconditionally (`InputBatch.java:43-44`). Running `ParquetDFSSource` at table version 9 and at table version 6 produced `streamer.checkpoint.key.v2` in **both** cases, so a DFS checkpoint does not stay on the `deltastreamer.*` key. The bullet no longer claims a table-version rule: > `--checkpoint` will set the matching reset key in the commit file to overwrite the current checkpoint, either `streamer.checkpoint.reset.key.v2` or `deltastreamer.checkpoint.reset_key`. Worth keeping in mind that the pre-existing text said `streamer.checkpoint.reset_key`, which is not a key Hudi writes at all — so this bullet needed correcting regardless of which rule governs the choice. -- 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]
