卡夫卡消费者民意调查无限期运行并且不返回任何内容答案

【问题标题】：Kafka Consumer Poll runs indefinitely and doesn't return anything卡夫卡消费者民意调查无限期运行并且不返回任何内容
【发布时间】：2020-05-25 08:44:08
【问题描述】：

我在使用 KafkaConsumer.poll(duration timeout) 时遇到困难，其中它无限期地运行并且永远不会退出该方法。了解这可能与连接有关，有时我看到它有点不一致。如果 poll 停止响应，我该如何处理？下面给出的是来自 KafkaConsumer.poll()

的 sn-p

public ConsumerRecords<K, V> poll(final Duration timeout) {
    return poll(time.timer(timeout), true);
}

我从这里调用上述内容：

Duration timeout = Duration.ofSeconds(30);
    while (true) {
        final ConsumerRecords<recordID, topicName> records = consumer.poll(timeout);
        System.out.println("record count is" + records.count());
}

我收到以下错误：

org.apache.kafka.common.errors.SerializationException: 错误在偏移量 2 处反序列化分区的键/值。如果有需要，请寻找过去的记录继续消费。

【问题讨论】：

如果它确实永远不会返回，您能否获取线程转储并查看您的线程在哪里？
请出示您的消费者财产的minimal reproducible example
我在上面添加了线程转储和我得到的错误

标签： apache-kafka kafka-consumer-api producer-consumer consumer

【解决方案1】：

在尝试解决上述问题时，我偶然发现了一些有用的信息。我将提供一段应该能够处理此问题的代码，但在此之前了解导致此问题的原因很重要。

在向 Apache Kafka 生成或使用消息或数据时，我们需要该消息或数据的模式结构，在我的例子中是 Avro 模式。如果向 Kafka 生成的消息与该消息模式冲突，则会对消费产生影响。

在消费记录的方法中，在消费者主题中添加以下代码 --

记得导入以下包：

导入 org.apache.kafka.common.TopicPartition;
导入 org.jsoup.SerializationException;

try {
        while (true) {
            ConsumerRecords<String, GenericRecord> records = null;
            try {
                records = consumer.poll(10000);
            } catch (SerializationException e) {
                String s = e.getMessage().split("Error deserializing key/value 
for partition ")[1].split(". If needed, please seek past the record to 
continue consumption.")[0];
                String topics = s.split("-")[0];
                int offset = Integer.valueOf(s.split("offset ")[1]);
                int partition = Integer.valueOf(s.split("-")[1].split(" at") . 
   [0]);

                TopicPartition topicPartition = new TopicPartition(topics, 
 partition);
                //log.info("Skipping " + topic + "-" + partition + " offset " 
 + offset);
                consumer.seek(topicPartition, offset + 1);
            }


            for (ConsumerRecord<String, GenericRecord> record : records) {

                System.out.printf("value = %s \n", record.value());


            }

        }


    } finally {
        consumer.close();
    }

【讨论】：

【解决方案2】：

我在设置测试环境时遇到了这个问题。

在代理上运行以下命令，如预期的那样打印出存储的记录：

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

原来是Kafka服务器配置错误。从外部连接 IP 地址listeners 必须在kafka/config/server.properties 中具有有效值，例如

# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://:9092

【讨论】：