fhan688 commented on code in PR #20486:
URL: https://github.com/apache/hudi/pull/20486#discussion_r4438543985
##########
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:
Agreed. `OffsetComparisonResult` now tracks whether each checkpoint actually
contains partition offsets, and `isExact()` requires offsets to be present on
both sides. Therefore, two Flink cluster-only checkpoints are treated as
inexact and full offset validation is skipped. Regression coverage was added at
the utility, Flink parsing, and validator levels.
--
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]
