### Question:
ProcessingKafkaConsumer implements Closeable { public void subscribe(Collection<String> topics) { consumer.subscribe(topics, rebalanceListener); } ProcessingKafkaConsumer(ProcessingConfig config);  ProcessingKafkaConsumer(ProcessingConfig config, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer);   ProcessingKafkaConsumer(ProcessingConfig config, Consumer<K, V> consumer); Optional<ConsumerRecord<K, V>> nextRecord(final long timeout); boolean ack(ConsumerRecord<K, V> record); boolean ack(TopicPartition topicPartition, long offset); boolean fail(ConsumerRecord<K, V> record); boolean fail(TopicPartition topicPartition, long offset); synchronized void resetOffsets(Map<TopicPartition, Long> offsets); synchronized void commitOffsets(); void subscribe(Collection<String> topics); Consumer<K, V> getConsumer(); @Override void close();  }### Answer:
@Test public void subscribe() { List<String> topics = Arrays.asList("topic1", "topic2"); processingConsumer.subscribe(topics); verify(consumer).subscribe(topics, processingConsumer.rebalanceListener); }