wombatu-kun commented on code in PR #20486:
URL: https://github.com/apache/hudi/pull/20486#discussion_r4434668096
##########
hudi-common/src/main/java/org/apache/hudi/common/util/KafkaCheckpointUtils.java:
##########
@@ -56,6 +59,54 @@
@Slf4j
public class KafkaCheckpointUtils {
+ /**
+ * Result of comparing offsets from two checkpoints.
+ *
+ * <p>The known offset difference only includes partitions whose offsets can
be compared
+ * exactly. A comparison is incomplete when partitions were added or
removed, or when an
+ * offset reset is detected, since the checkpoints do not contain the start
offset used for
+ * those partitions.</p>
+ */
+ public static final class OffsetComparisonResult {
+ private final long knownOffsetDifference;
+ private final Set<Integer> addedPartitions;
+ private final Set<Integer> resetPartitions;
+ private final Set<Integer> missingPartitions;
+
+ private OffsetComparisonResult(long knownOffsetDifference,
+ Set<Integer> addedPartitions,
+ Set<Integer> resetPartitions,
+ Set<Integer> missingPartitions) {
+ this.knownOffsetDifference = knownOffsetDifference;
+ this.addedPartitions = Collections.unmodifiableSet(new
TreeSet<>(addedPartitions));
+ this.resetPartitions = Collections.unmodifiableSet(new
TreeSet<>(resetPartitions));
+ this.missingPartitions = Collections.unmodifiableSet(new
TreeSet<>(missingPartitions));
+ }
+
+ public long getKnownOffsetDifference() {
+ return knownOffsetDifference;
+ }
+
+ public Set<Integer> getAddedPartitions() {
+ return addedPartitions;
+ }
+
+ public Set<Integer> getResetPartitions() {
+ return resetPartitions;
+ }
+
+ public Set<Integer> getMissingPartitions() {
+ return missingPartitions;
+ }
+
+ /**
+ * Returns whether the offset difference covers all partitions in both
checkpoints.
+ */
+ public boolean isExact() {
+ return addedPartitions.isEmpty() && resetPartitions.isEmpty() &&
missingPartitions.isEmpty();
Review Comment:
When the Flink offset lookup fails open on two commits in a row,
`StreamerUtil.collectKafkaOffsetCheckpoint` writes only the `kafka_cluster`
entry both times, so both sides parse to no partitions, `isExact` returns true,
and a difference of 0 against the full write count still rejects the commit.
Should a comparison with no partitions on either side also count as inexact?
##########
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:
This skip also gives up what the known part still proves: with `topic,0:100`
-> `topic,0:110,1:5` and 3 records written, at least 7 records are gone
whatever partition 1 contributed, yet that commit now passes where it used to
be rejected. Is that intended, or should the inconclusive branch still fail
when `recordsWritten` is below the known difference beyond the tolerance?
##########
hudi-common/src/main/java/org/apache/hudi/common/util/KafkaCheckpointUtils.java:
##########
@@ -109,28 +160,61 @@ public static Map<Integer, Long>
parseCheckpoint(CheckpointFormat format, String
}
/**
- * Calculate offset difference between two checkpoints.
- * Handles partition additions, removals, and resets.
+ * Calculate the known offset difference between two checkpoints.
*
* Algorithm:
* 1. For each partition in current checkpoint:
* - If partition exists in previous: diff = current - previous
* - If partition is new: skip (start offset unknown, would overcount)
* - If diff is negative (reset): skip (start offset unknown, would
overcount)
- * 2. Sum all partition diffs
+ * 2. Sum all comparable partition diffs
+ *
+ * <p>If partitions are added or removed, or an offset reset is detected,
the returned value
+ * is only a partial difference. Callers that require an exact difference
should use
+ * {@link #compareOffsets(CheckpointFormat, String, String)} and check
+ * {@link OffsetComparisonResult#isExact()}.</p>
*
* @param format Checkpoint format
* @param previousCheckpoint Previous checkpoint string
* @param currentCheckpoint Current checkpoint string
- * @return Total offset difference across all partitions
+ * @return Known offset difference across comparable partitions
*/
public static long calculateOffsetDifference(CheckpointFormat format,
Review Comment:
`calculateOffsetDifference` has no production caller left, only tests, and
it still returns the partial difference this PR stops the validator from
trusting. Can we drop it and move its Spark and Flink tests to
`compareOffsets`, or is there an external consumer to keep it for?
--
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]
