Estoy tratando de hacer que un Kafka Consumer simple funcione con la API de Java v0.9.0.1. El servidor kafka que estoy usando es un contenedor acoplable, que también ejecuta la versión 0.9.0.1. A continuación se muestra el código del consumidor:

public class Consumer {
    public static void main(String[] args) throws IOException {

        KafkaConsumer<String, String> consumer;
        try (InputStream props = Resources.getResource("consumer.props").openStream()) {
            Properties properties = new Properties();
            properties.load(props);
            consumer = new KafkaConsumer<>(properties);
        }

        consumer.subscribe(Arrays.asList("messages"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records)
                    System.out.println("Message received: " + record.value());
            }
        }catch(WakeupException ex){
            System.out.println("Exception caught " + ex.getMessage());
        }finally{
            consumer.close();
            System.out.println("After closing KafkaConsumer");
        }
    }
}

Sin embargo, al iniciar al consumidor, llama al método poll (100) anterior y nunca regresa. Depuración, parece que se atasca ejecutando el siguiente método en org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient para siempre:

public void awaitMetadataUpdate() {
    int version = this.metadata.requestUpdate();

    do {
        this.poll(9223372036854775807L);
    } while(this.metadata.version() == version);

}

(tanto la versión como this.metadata.version () siempre parecen ser == 2). Además, aunque no arroja errores, los mensajes de mi productor java nunca llegaron a la cola. Verifiqué que usando las herramientas de línea de comando kafka, puedo enviar y recibir mensajes de la cola.

Alguien tiene alguna idea de lo que está pasando aquí?