fhan688 commented on code in PR #20486:
URL: https://github.com/apache/hudi/pull/20486#discussion_r4438540475
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/validator/StreamingOffsetValidator.java:
##########
@@ -153,11 +154,18 @@ public void validateWithMetadata(ValidationContext
context) throws HoodieValidat
return;
}
- // Calculate offset difference using format-specific logic.
- // Note: calculateOffsetDifference always returns >= 0 because negative
per-partition
- // diffs (offset resets) are handled internally by using the current
offset.
- long offsetDifference = KafkaCheckpointUtils.calculateOffsetDifference(
+ // Calculate the offset difference using format-specific logic.
Checkpoint-only comparison
+ // cannot determine the consumed record count for added, reset, or missing
partitions.
+ OffsetComparisonResult offsetComparison =
KafkaCheckpointUtils.compareOffsets(
checkpointFormat, previousCheckpoint, currentCheckpoint);
+ if (!offsetComparison.isExact()) {
Review Comment:
Good catch. I’ve updated the incomplete-comparison path to preserve the
known offset difference as a lower bound. If `recordsWritten` is below that
bound, the validator now applies the configured tolerance and `FAIL/WARN_LOG`
policy. When `recordsWritten` is at or above the bound, the full comparison
remains inconclusive because the unknown partitions may contribute additional
records. I also added regression tests for strict failure, tolerance, and
`WARN_LOG` behavior.
--
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]
