时间戳

首页 > 解决方案 > Kafka 客户端 poll() 每次收到消息后都会抛出 EOFError

apache-kafka - Kafka 客户端 poll() 每次收到消息后都会抛出 EOFError

问题描述

我正在使用我在Confluent github中找到的示例客户端轮询代码的一个轻微变体：

c = Consumer({'bootstrap.servers':'localhost:9092','group.id':'devops','auto.offset.reset':'earliest'})
c.subscribe(['system-diskio-write-bytes','system-cpu-user-pct'])

try:
    while True:
        msg = c.poll(timeout=1000.0)
        if msg is None:
            continue
        if msg.error():
            print(msg.error())
        else:
            print('topic: %s key: %s value: %s' % (msg.topic(), msg.key(), msg.value()))

except KeyboardInterrupt:
    print('Polling interrupted by consumer')

在收到每条消息后都会引发 EOF KafkaError：

topic: system-diskio-write-bytes key: None value: b'{"route" : "system-diskio-write-bytes", "timestamp" : 2019-03-06T13:46:25.244, "value" : 655002980352.0}'
KafkaError{code=_PARTITION_EOF,val=-191,str="Broker: No more messages"}

我不明白为什么会发生这种情况 - 关于为什么抛出这个错误的任何想法如何解决？任何想法都非常感谢 - 谢谢！

标签： apache-kafkakafka-consumer-api

解决方案

好的，这里详细解答。解释如下：

无论您的应用程序是否在 10 秒内没有调用 poll()，每次消费者遇到新的结束偏移量时，都会将 EOF 事件推送到内部消息队列（由 poll() 提供服务）。

可以按如下方式禁用 EOF 事件：

enable.partition.eof=false

推荐阅读