【问题标题】:Why is my consumer reading all messages from the topic each time, even when auto.offset.reset=largest?为什么我的消费者每次都读取来自该主题的所有消息,即使 auto.offset.reset=largest?
【发布时间】:2017-07-07 06:23:45
【问题描述】:

我在 topic1 上向 Kafka 生成了 5 条消息并成功使用它们。当我发送第 6 条消息并尝试使用时,我再次收到所有 6 条消息,而不是最新的(第 6 条)消息。

请注意我正在运行消费者命令行,而不是从数据库连接器(访问模块)。并且连接器的配置属性 auto.offset.reset 设置为“最大”。(请参阅下面日志中的所有配置属性)

另请参阅下面的 OffsetChecker 输出:

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker \
    --group testjob --zookeeper localhost:2181 --topic topic1

[2017-07-06 21:57:46,707] WARN WARNING: ConsumerOffsetChecker is deprecated and will be dropped in releases following 0.9.0. Use ConsumerGroupCommand instead. (kafka.tools.ConsumerOffsetChecker$)
Exiting due to: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /consumers/testjob/offsets/topic1/0.

谁能告诉我问题出在哪里?

这是显示配置属性的日志:

***Global config Properties***
*             client.id = rdkafka
*             message.max.bytes = 1200
*             receive.message.max.bytes = 100000000
*             metadata.request.timeout.ms = 60000
*             topic.metadata.refresh.interval.ms = 600000
*             topic.metadata.refresh.fast.cnt = 10
*             topic.metadata.refresh.fast.interval.ms = 250
*             topic.metadata.refresh.sparse = false
*             socket.timeout.ms = 60000
*             socket.send.buffer.bytes = 0
*             socket.receive.buffer.bytes = 0
*             socket.keepalive.enable = false
*             socket.max.fails = 3
*             broker.address.ttl = 300000
*             broker.address.family = any
*             statistics.interval.ms = 0
*             log_cb = 0x7fecb80c6dd0
*             log_level = 6
*             socket_cb = 0x7fecb80cd2f0
*             open_cb = 0x7fecb80ddd30
*             opaque = 0x2641280
*             internal.termination.signal = 0
*             queued.min.messages = 100000
*             queued.max.messages.kbytes = 1000000
*             fetch.wait.max.ms = 100
*             fetch.message.max.bytes = 1049776
*             fetch.min.bytes = 1
*             fetch.error.backoff.ms = 500
*             group.id = testjob
*             queue.buffering.max.messages = 100000
*             queue.buffering.max.ms = 1000
*             message.send.max.retries = 2
*             retry.backoff.ms = 100
*             compression.codec = none
*             batch.num.messages = 1000
*             delivery.report.only.error = false
*             request.required.acks = 1
*             enforce.isr.cnt = 0
*             request.timeout.ms = 5000
*             message.timeout.ms = 300000
*             produce.offset.report = false
*             auto.commit.enable = true
*             auto.commit.interval.ms = 60000
*             auto.offset.reset = largest    <<<<--------
*             offset.store.path = .
*             offset.store.sync.interval.ms = 0
*             offset.store.method = file
*             consume.callback.max.messages = 0

【问题讨论】:

  • 你是如何运行消费者的?一个完整的命令行可能有助于诊断出了什么问题。
  • 在不了解您的消费者的情况下,只能猜测。也许您的消费者运行不到 60 秒(auto.commit.interval)并且被杀死而不是正常关闭。关于 zookeeper 上缺少的节点:可能是您正在运行一个“新消费者”,它没有将偏移量提交给 ZK。或者您不写入 ZK 的根路径(我会推荐)。检查您的代理配置(zookeeper.connect)。这可能看起来像:'localhost:2181/kafka' - 在这种情况下,您必须在运行偏移检查器时将路径添加到 ZK 连接字符串。

标签: apache-kafka offset kafka-consumer-api


【解决方案1】:

添加此属性 AUTO_OFFSET_RESET_CONFIG="earliest" 它会工作

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2019-12-04
    • 2021-07-09
    • 1970-01-01
    • 1970-01-01
    • 2020-12-12
    • 2020-09-19
    • 2019-06-11
    相关资源
    最近更新 更多