- Map<Integer, Long> offsets = new HashMap<>();
- doForCurrentOffsets((tp, offset) -> offsets.put(tp.partition(), offset -1));
- Consumer<ConsumerRecord<String, Long>> captureOffset = record -> offsets.put(record.partition(), record.offset());
+ Map<TopicPartition, Long> offsets = new HashMap<>();
+ doForCurrentOffsets((tp, offset) -> offsets.put(tp, offset -1));
+ Consumer<ConsumerRecord<String, Long>> captureOffset =
+ record ->
+ offsets.put(
+ new TopicPartition(record.topic(), record.partition()),
+ record.offset());