卡夫卡消费者没有正确提交偏移量答案

【问题标题】：Kafka consumer not committing offset correctly卡夫卡消费者没有正确提交偏移量
【发布时间】：2019-12-15 10:08:40
【问题描述】：

我有一个使用以下属性定义的 Kafka 消费者：

session.timeout.ms = 60000
heartbeat.interval.ms = 6000

我们注意到大约 2000 条消息的延迟，并看到消费者多次使用同一条消息（通过我们的应用日志）。此外，注意到一些消息需要大约 10 秒才能完全处理。我们的怀疑是消费者没有正确提交偏移量（或重复提交相同的旧偏移量），因此消费者收到了相同的消息。

为了解决这个问题，我们引入了更多属性：

auto.commit.interval.ms=20000 //To ensure that commit is happening only after processing of message is completed
max.poll.records=10 //To make the consumer pick only 10 messages in one go

And, we set the concurrency to 1.

这解决了我们的问题。滞后开始减少，最终为 0。

但是，我仍然不清楚为什么会首先出现问题。据我了解，默认情况下：

enable.auto.commit = true
auto.commit.interval.ms=5000

因此，理想情况下，消费者应该每 5 秒提交一次。如果在此时间范围内没有完全处理消息，会发生什么？消费者正在提交什么补偿？问题是否是由于投票记录大小过大（默认为 500）引起的

另外，关于 poll() 方法，我读到：

poll() 调用在设置的 auto.commit.interval.ms 的后台发出。

那么，最初如果 poll() 早于每 5 秒发生一次（默认为 auto.commit.interval），为什么它不提交最新的偏移量？因为消费者还没有做完处理吗？然后，它应该在接下来的 5 秒提交该偏移量。

有人可以回答这些问题并解释为什么会出现最初的问题吗？

【问题讨论】：

标签： spring spring-boot apache-kafka kafka-consumer-api spring-kafka

【解决方案1】：

如果您使用 Spring for Apache Kafka，我们建议将 enable.auto.commit 设置为 false 以便容器以更确定的方式提交偏移量（在每条记录之后，或每批记录 - 默认） .

问题很可能是max.poll.interval.ms，默认为 5 分钟。如果您的一批消息花费的时间比这更长，您就会看到这种行为。您可以增加max.poll.interval.ms，或者像您所做的那样减少max.poll.records。

关键是您必须在小于max.poll.interval.ms 的时间内处理民意调查返回的记录。

另外，关于 poll() 方法，我读到：

poll() 调用在设置的 auto.commit.interval.ms 的后台发出。

这是不正确的； poll() 不会在后台调用；自 KIP-62 起，心跳在后台发送。

【讨论】：

谢谢。这是有道理的，因为只有在我们通过 cron 将 1700 条消息推送到主题后，问题才开始出现。在此之前，该主题的消息从未真正超过 100/200 条。
@Gary Russell，如果下一次 poll() 超过 5 秒（例如，处理时间超过 5 秒），kafka 消费者会等待还是会发出 poll 方法？（我怀疑它不应该，它会等待，因此不尊重 poll() 方法，在这种情况下，提交只会在处理完所有记录后发生，这些记录是从当前 poll() 方法中检索出来的）——因此与此语句相矛盾- “poll() 调用是在后台设置的 auto.commit.interval.ms 发出的。”
不要在 cmets 中问补充问题。是的，它与该声明相矛盾，因为poll() 没有在后台调用。仅在后台发送心跳（自 KIP-62 起）。当使用自动提交时（不推荐在 Spring 中使用），提交只会发生（在下一次调用 poll() 期间），如果间隔已过；如果间隔没有过去，提交将在随后的轮询之前发生。对于 spring，使用 BATCH（默认）或 RECORD AckMode 意味着提交将更加确定地发生，与时间无关。我们现在将 enable.auto.commit 默认设置为 false。