KafkaConsumer.position should wait for assignment metadata

consumer.subscribe(topics);
// consumer.poll(0); <- I've removed this line, which shouldn't be necessary here.

final Set<TopicPartition> partitions = new HashSet<>();
for (final String topic : topics) {
    for (final PartitionInfo partition : consumer.partitionsFor(topic)) {
        partitions.add(new TopicPartition(partition.topic(), partition.partition()));
    }
}

for (final TopicPartition tp : partitions) {
    final long offset = consumer.position(tp);
    committedOffsets.put(tp, offset);
}

Exception in thread "main" java.lang.IllegalStateException: You can only check the position for partitions assigned to this consumer.
   at org.apache.kafka.clients.consumer.KafkaConsumer.position(KafkaConsumer.java:1620)
   at org.apache.kafka.clients.consumer.KafkaConsumer.position(KafkaConsumer.java:1586)
   at org.apache.kafka.streams.tests.EosTestDriver.getCommittedOffsets(EosTestDriver.java:275)
   at org.apache.kafka.streams.tests.EosTestDriver.verify(EosTestDriver.java:148)
   at org.apache.kafka.streams.tests.StreamsEosTest.main(StreamsEosTest.java:69)