yihua commented on code in PR #10875:
URL: https://github.com/apache/hudi/pull/10875#discussion_r1545142959
##########
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:
Generally, it's a bad pattern to filter and change the same map
`finalRanges` in the Java stream operation. You may collect the items into a
list after filtering and then do foreach to avoid any unexpected behavior.
--
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]
