kafka 消费者 API consumer.poll() 不能正常工作，没有错误，没有异常，只是阻塞

Question

我正在按照 Apache kafka 文档学习 kafka。我使用默认配置启动它。

bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties &

我运行了 kafka-console-producer.sh 和 kafka-console-consumer.sh 来生产和消费消息，并且成功了。我使用生产者 API 编写了一个 java 代码来生成消息，这没关系。这由 kafka-console-consumer.sh 验证。代码与 Apache Kafka 指南相同：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer",    "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer<String, String> producer = new KafkaProducer<>(props);
for(int i = 0; i < 100; i++)
  producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));

producer.close();

虽然生产者代码有效，但消费者代码无效。没有例外，但它只是在 consumer.poll(100) 处阻塞。代码来自 Apache Kafka 文档：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records)
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
 }

顺便说一句，Apache Kafka 文档中的 kafka-console-consumer.sh 示例消费消息成功，由 producer 生产到 topic “test”：

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

但是如果我不连接zookeeper，而是直接连接kafka broker，那么它也不能正常工作，没有错误，没有异常，它只是阻塞。

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

kafka版本和API版本为0.11.0.0

为什么他们不能消费消息？

Answer 1

使用--zookeeper 参数意味着使用旧的消费者，它运行良好，因为您指定了一个Zookeeper 服务器（localhost:2181）。

当您想指定一个 Kafka 代理（因此使用新的消费者）时，您必须使用 --bootstrap-server 选项：您仍在使用 --zookeeper，但传递了一个有效的 Kafka 代理地址 (localhost:9092)。

因此，对于控制台使用者应用程序，您的配置需要为 --bootstrap-server localhost:9092 而不是 --zookeeper localhost:9092。

关于您的代码，您确定 poll 方法被阻止了吗？如果没有记录但没有阻塞，它应该在 100 毫秒（您指定的超时）后退出。

然后从您的代码中我看到生产者正在发送到“my-topic”，消费者订阅了“foo”和“bar”；最后，控制台消费者从“测试”中读取。都是不同的话题！

Answer 2

props.put("auto.offset.reset", "smallest"); 添加此属性，希望它可以工作