@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));
}