### Question:
ConsumerOffsetClient implements Closeable { public Collection<TopicPartition> getPartitionsFor(Collection<String> topics) { if (topics == null) throw new IllegalArgumentException("topics cannot be null"); return topics.stream() .flatMap(topic -> { List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic); if (partitionInfos == null) return Collections.<TopicPartition>emptyList().stream(); return partitionInfos.stream() .map(partitionInfo -> new TopicPartition(partitionInfo.topic(), partitionInfo.partition())); }) .collect(Collectors.toList()); } 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 getPartitionsFor() { 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))); assertThat(client.getPartitionsFor(Arrays.asList("topic1", "topic2")), is(Arrays.asList( new TopicPartition("topic1", 0), new TopicPartition("topic1", 1), new TopicPartition("topic2", 0), new TopicPartition("topic2", 1) ))); }
@Test (expected = IllegalArgumentException.class) public void getPartitionsFor_nullTopics() { client.getPartitionsFor(null); }