bvaradar commented on code in PR #10875:
URL: https://github.com/apache/hudi/pull/10875#discussion_r1545549852
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java:
##########
@@ -174,6 +174,13 @@ public static OffsetRange[]
computeOffsetRanges(Map<TopicPartition, Long> fromOf
}
}
}
+ // We need to ensure every partition is part of returned offset ranges
even if we are not consuming any new msgs (for instance, if its already caught
up).
+ // as this will be tracked as the checkpoint, we need to ensure all
partitions are part of final ranges.
+ fromOffsetMap.entrySet()
+ .stream()
+ .filter((kv) -> !finalRanges.containsKey(kv.getKey()))
+ .forEach((kv) -> finalRanges.put(kv.getKey(),
Collections.singletonList(OffsetRange.create(kv.getKey(), kv.getValue(),
kv.getValue()))));
Review Comment:
+1
--
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]
