Consumidor Java Kafka 0.9.0.1 preso em waititMetadataUpdate ()
Estou tentando fazer com que um simples consumidor Kafka funcione usando a API Java v0.9.0.1. O servidor kafka que estou usando é um contêiner de docker, também executando a versão 0.9.0.1. Abaixo está o código do 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");
}
}
}
No entanto, ao iniciar o consumidor, ele chama o método poll (100) acima e nunca retorna. Na depuração, parece que ele fica preso ao executar o seguinte método em org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient para sempre:
public void awaitMetadataUpdate() {
int version = this.metadata.requestUpdate();
do {
this.poll(9223372036854775807L);
} while(this.metadata.version() == version);
}
(a versão e this.metadata.version () sempre parecem ser == 2). Além disso, apesar de não gerar erros, as mensagens do meu produtor java nunca foram vistas para chegar à fila. Eu verifiquei que, usando as ferramentas kafka da linha de comando, posso enviar e receber mensagens da fila.
Alguém tem alguma idéia do que está acontecendo aqui?