fhan688 commented on code in PR #20486:
URL: https://github.com/apache/hudi/pull/20486#discussion_r4451666448
##########
hudi-common/src/main/java/org/apache/hudi/common/util/KafkaCheckpointUtils.java:
##########
@@ -109,28 +176,28 @@ public static Map<Integer, Long>
parseCheckpoint(CheckpointFormat format, String
}
/**
- * Calculate offset difference between two checkpoints.
- * Handles partition additions, removals, and resets.
+ * Compare two checkpoints while preserving whether the resulting offset
difference is exact.
*
- * 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
+ * <p>Offsets for partitions present in both checkpoints are included when
they move forward or
+ * remain unchanged. Partitions with an unknown start offset are reported
separately and excluded
+ * from the known difference.</p>
*
* @param format Checkpoint format
* @param previousCheckpoint Previous checkpoint string
* @param currentCheckpoint Current checkpoint string
- * @return Total offset difference across all partitions
+ * @return Offset comparison result containing the known difference and
incomparable partitions
*/
- public static long calculateOffsetDifference(CheckpointFormat format,
- String previousCheckpoint,
- String currentCheckpoint) {
+ public static OffsetComparisonResult compareOffsets(CheckpointFormat format,
+ String
previousCheckpoint,
+ String
currentCheckpoint) {
Map<Integer, Long> previousOffsets = parseCheckpoint(format,
previousCheckpoint);
Map<Integer, Long> currentOffsets = parseCheckpoint(format,
currentCheckpoint);
long totalDiff = 0;
+ Set<Integer> addedPartitions = new TreeSet<>();
+ Set<Integer> resetPartitions = new TreeSet<>();
+ Set<Integer> missingPartitions = new TreeSet<>(previousOffsets.keySet());
+ missingPartitions.removeAll(currentOffsets.keySet());
Review Comment:
Good suggestion. The common case has identical partition sets, so
materializing all previous partitions in a `TreeSet` only to remove them is
unnecessary. I’ll change this to scan `previousOffsets` and add only partitions
absent from `currentOffsets`, preserving sorted diagnostics while avoiding the
temporary `O(P)` tree allocation.
##########
hudi-common/src/main/java/org/apache/hudi/common/util/KafkaCheckpointUtils.java:
##########
@@ -56,6 +59,70 @@
@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 either checkpoint has no
partition offsets,
+ * partitions were added or removed, or an offset reset is detected, since
the checkpoints
+ * do not contain enough information to calculate the full offset
difference.</p>
+ */
+ public static final class OffsetComparisonResult {
+ private final long knownOffsetDifference;
+ private final boolean previousCheckpointHasOffsets;
+ private final boolean currentCheckpointHasOffsets;
+ private final Set<Integer> addedPartitions;
+ private final Set<Integer> resetPartitions;
+ private final Set<Integer> missingPartitions;
+
+ private OffsetComparisonResult(long knownOffsetDifference,
+ boolean previousCheckpointHasOffsets,
+ boolean currentCheckpointHasOffsets,
+ Set<Integer> addedPartitions,
+ Set<Integer> resetPartitions,
+ Set<Integer> missingPartitions) {
+ this.knownOffsetDifference = knownOffsetDifference;
+ this.previousCheckpointHasOffsets = previousCheckpointHasOffsets;
+ this.currentCheckpointHasOffsets = currentCheckpointHasOffsets;
+ this.addedPartitions = Collections.unmodifiableSet(new
TreeSet<>(addedPartitions));
+ this.resetPartitions = Collections.unmodifiableSet(new
TreeSet<>(resetPartitions));
+ this.missingPartitions = Collections.unmodifiableSet(new
TreeSet<>(missingPartitions));
Review Comment:
Agreed. The sets are created locally by `compareOffsets()`, ownership is
transferred to the private result constructor, and they are never mutated
afterward. I’ll wrap the supplied `TreeSets` directly with
`Collections.unmodifiableSet(...)` to retain the immutable public view without
copying all entries again.
--
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]
