### Question:
ConsumerOffsetClient implements Closeable { public Map<TopicPartition, Long> getCommittedOffsets(Collection<String> topics) { if (topics == null) throw new IllegalArgumentException("topics cannot be null"); Collection<TopicPartition> partitions = getPartitionsFor(topics); Map<TopicPartition, Long> offsets = new HashMap<>(); partitions.forEach(topicPartition -> { offsets.put(topicPartition, getCommittedOffset(topicPartition)); }); return offsets; } ConsumerOffsetClient(Properties properties);  ConsumerOffsetClient(Consumer<Object, Object> consumer);   ConsumerOffsetClient(Consumer<Object, Object> consumer, boolean providedByUser); Map<TopicPartition, Long> getEndOffsets(Collection<String> topics); Map<TopicPartition, Long> getBeginningOffsets(Collection<String> topics); Map<TopicPartition, Long> getOffsetsForTimes(Collection<String> topics, long time); Map<TopicPartition, Long> getCommittedOffsets(Collection<String> topics); void commitOffsets(Map<TopicPartition, Long> offsets); long getCommittedOffset(TopicPartition topicPartition); Collection<TopicPartition> getPartitionsFor(Collection<String> topics); @Override void close();  }### Answer:
@Test public void getCommittedOffsets() { Map<TopicPartition, Long> offsets = new HashMap<>(); offsets.put(new TopicPartition("topic1", 0), 123L); offsets.put(new TopicPartition("topic1", 1), 234L); offsets.put(new TopicPartition("topic2", 0), -1L); offsets.put(new TopicPartition("topic2", 1), -1L); when(consumer.partitionsFor("topic1")).thenReturn(Arrays.asList( new PartitionInfo("topic1", 0, null, null, null), new PartitionInfo("topic1", 1, null, null, null))); when(consumer.partitionsFor("topic2")).thenReturn(Arrays.asList( new PartitionInfo("topic2", 0, null, null, null), new PartitionInfo("topic2", 1, null, null, null))); when(consumer.committed(new TopicPartition("topic1", 0))).thenReturn(new OffsetAndMetadata(123L)); when(consumer.committed(new TopicPartition("topic1", 1))).thenReturn(new OffsetAndMetadata(234L)); assertThat(client.getCommittedOffsets(Arrays.asList("topic1", "topic2")), is(offsets)); }
@Test (expected = IllegalArgumentException.class) public void getCommittedOffsets_nullTopics() { client.getCommittedOffsets(null); }