danny0405 commented on code in PR #20486:
URL: https://github.com/apache/hudi/pull/20486#discussion_r4450431090
##########
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:
Could we iterate previousOffsets.keySet() and add only partitions absent
from currentOffsets to missingPartitions? Building a TreeSet of all previous
partitions adds O(P log P) work and P temporary tree nodes on every comparison,
even when the partition sets are identical and the result is empty. Collecting
only missing IDs would preserve sorted diagnostics while keeping the common
case linear. This is a small performance concern; end-to-end impact has not
been benchmarked.
##########
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:
These three defensive copies look unnecessary with the current ownership
model: this constructor is private, and its only caller creates the sets
locally and never mutates or exposes them afterward. Could we wrap the supplied
sets directly with Collections.unmodifiableSet(...) instead? That would retain
the immutable public view while avoiding a second set allocation and copying
the entries.
--
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]
