deepakpanda93 commented on code in PR #19573: URL: https://github.com/apache/hudi/pull/19573#discussion_r3750547072
########## 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. - `--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. +#### Resetting the checkpoint for the Hudi incremental source + +When Hudi Streamer writes a target table at table version 8 or higher using `HoodieIncrSource`, it tracks progress by +**completion time** instead of requested instant time, and records it in the commit metadata under +`streamer.checkpoint.key.v2`. A bare timestamp would be ambiguous between the two, so `--checkpoint` has to state which +one it is. This form is required from Hudi 1.0.1 onward; on 1.0.0 the prefixes are not recognised and `--checkpoint` +takes a bare completion time. + +```shell Review Comment: Good challenge — I re-tested rather than argue from the earlier run, and the doc statement holds: **1.0.0 reads a bare `--checkpoint` as a completion time.** The inference in the comment is that because the bare value *was accepted* on 1.0.0, and that value happened to be a requested instant time, 1.0.0 must interpret bare values as requested times. Acceptance does not imply interpretation, though — 1.0.0 does no validation of the value at all, so it accepts any timestamp-shaped string regardless of which clock it came from. What the value *means* only shows up in which records get ingested. So I ran the discriminating pair on 1.0.0. Source instant #2: requested `20260810143709423`, completed `20260810143710155`. | bare `--checkpoint` | rows ingested | instant #2 | |---|---|---| | `20260810143709423` (its **requested** time) | `21,22,31,32,41,42` | **included** | | `20260810143710155` (its **completion** time) | `31,32,41,42` | **excluded** | If a bare value were a requested time, passing instant #2's requested time would have resumed *after* instant #2 and excluded its records. It did not. Passing instant #2's **completion** time is what lands the checkpoint exactly after it. That is completion time semantics. This also explains the original matrix row that prompted the question. The bare value I fed every version was instant #2's requested time, which is *earlier* than its completion time, so read as a completion time it sits mid-way and instant #2 is still ahead of the checkpoint — hence six records on 1.0.0 where 1.0.1+ ingests four. The source agrees. In `release-1.0.0`, `StreamerCheckpointUtils` L67-68: ```java checkpoint = Option.of(CheckpointUtils.targetCheckpointV2(writeTableVersion) ? new StreamerCheckpointV2(streamerConfig.checkpoint) : new StreamerCheckpointV1(streamerConfig.checkpoint)); ``` At table version 8 `targetCheckpointV2` is true, so the raw string is wrapped in `StreamerCheckpointV2` — a completion time — with no translation and no parsing. `HoodieIncrSourceCheckpointValUtils`, which is what introduces the prefixes and their validation, does not exist in 1.0.0 (404 at that tag). On #12718's note that 1.x was "still using request time based handling": that was written in January 2025 and is now stale, which the PR body covers. The 1.0.0 run with no override stores `streamer.checkpoint.key.v2` holding a completion time, so even 1.0.0 was already completion time based in storage — it just had no way to disambiguate an override, which is the gap #12718 closed. No change made for this one. @yihua a confirmation of the pre-1.0.1 semantics would still be welcome. ########## 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. - `--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: Right that these do not reconcile — and the reason is that `streamer.checkpoint.reset_key` is not a real key either. Fixed in 05e5b6443aa2. It has the same missing-prefix defect as the checkpoint key line I corrected in d92a88e38e9c. The actual reset keys, at `release-1.2.0`: | Checkpoint | Reset key | Defined / written | |---|---|---| | request time | `deltastreamer.checkpoint.reset_key` | `StreamerCheckpointV1.java:30`, written at `:60` | | completion time | `streamer.checkpoint.reset.key.v2` | `StreamerCheckpointV2.java:33`, written at `:66` | Same `CheckpointUtils.shouldTargetCheckpointV2` test on the write table version that selects between `deltastreamer.checkpoint.key` and `streamer.checkpoint.key.v2`. So the bullet was not describing an older key superseded by the new one — it was naming a key Hudi never writes, which is exactly why a reader inspecting commit metadata would not find it. The bullet now reads: > `--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. which mirrors the phrasing used for the checkpoint key a few lines above, so the two lines now agree with each other and with the new subsection. Empirically confirmed earlier in this PR: the version-9 target commit carries `streamer.checkpoint.reset.key.v2` with the prefixed override verbatim, and a target forced to `hoodie.write.table.version=6` carries the `deltastreamer.*` pair. Build is clean with the warning set byte-identical to the baseline. -- 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]
