OffsetsByPartitionFunction { public Map<TopicPartition, Offsets> apply(Collection<TopicPartition> partitions) { Map<TopicPartition, Long> beginningOffsets; Map<TopicPartition, Long> endOffsets; synchronized (consumer) { beginningOffsets = consumer.beginningOffsets(partitions); endOffsets = consumer.endOffsets(partitions); } return Flux.fromIterable(partitions).collectMap(identity(), tp -> { long beginning = beginningOffsets.getOrDefault(tp, 0L); long end = endOffsets.getOrDefault(tp, 0L); return new Offsets(beginning, end); }).block(); }  Map<TopicPartition, Offsets> apply(Collection<TopicPartition> partitions);  }
@Test public void typical() throws Exception { TopicPartition topicPartition = new TopicPartition("topic", 0); Collection<TopicPartition> topicPartitions = singleton(topicPartition); doReturn(singletonMap(topicPartition, 1L)).when(kafkaConsumer).beginningOffsets(topicPartitions); doReturn(singletonMap(topicPartition, 3L)).when(kafkaConsumer).endOffsets(topicPartitions); Map<TopicPartition, Offsets> result = underTest.apply(topicPartitions); assertThat(result.size(), is(1)); Offsets offsets = result.get(topicPartition); assertThat(offsets.getBeginning(), is(1L)); assertThat(offsets.getEnd(), is(3L)); assertThat(offsets.getCount(), is(2L)); }