- return
- IntStream
- .range(0, PARTITIONS)
- .mapToObj(partition -> new TopicPartition(TOPIC, partition))
- .collect(Collectors.toList());
- }
-
- void doForCurrentOffsets(BiConsumer<TopicPartition, Long> consumer)
- {
- kafkaConsumer.assign(partitions());
- partitions().forEach(tp -> consumer.accept(tp, kafkaConsumer.position(tp)));
- kafkaConsumer.unsubscribe();
- }
-
- void checkSeenOffsetsForProgress()
- {
- // Be sure, that some messages were consumed...!
- Set<TopicPartition> withProgress = new HashSet<>();
- partitions().forEach(tp ->