卡夫卡抛出“org.apache.kafka.clients.consumer.CommitFailedException”答案

【问题标题】：Kafka throwing "org.apache.kafka.clients.consumer.CommitFailedException"卡夫卡抛出“org.apache.kafka.clients.consumer.CommitFailedException”
【发布时间】：2019-09-27 13:15:26
【问题描述】：

我使用 spring-kafka 库开发了 Kafka 消费者应用程序，并使用了手动提交的默认消费者配置。

我正在运行两个应用程序实例，监听两个不同的 Kafka 主题。在执行负载测试时，我观察到我只有一个应用程序的负载低于错误：

Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. 

    This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, 
    which typically implies that the poll loop is spending too much time message processing. 

    You can address this either by increasing the session timeout or by reducing the maximum size of batches 
    returned in poll() with max.poll.records.


org.apache.kafka.clients.consumer.CommitFailedException: 
    Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. 

    This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, 
    which typically implies that the poll loop is spending too much time message processing. 


You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.
    \n org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:725)

我阅读了几篇文章，发现如果消费者花费大量时间处理消息并且代理没有获得有关消费者活跃度的信息，那么消费者重新平衡就会发生，并且对于未提交的消息将抛出上述异常。

我已通过将 max.poll.interval.ms 设置为 INEGER.MAX_VALUE 解决了上述错误。但我想知道为什么我只在一个实例中出现错误，为什么其他实例在更高负载下按预期工作。

任何人都可以分享max.poll.interval.ms 的正确根本原因和理想值或此问题的适当解决方案

【问题讨论】：

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

【解决方案1】：

除了 Yoav 的建议之外，如果减小批量大小不是一个选项，您还可以尝试增加 max.poll.interval.ms 的值。来自 Kafka 文档：

使用消费者时调用 poll() 之间的最大延迟群组管理。这为时间量设置了上限消费者可以在获取更多记录之前处于空闲状态。如果轮询（）在此超时到期之前不调用，那么消费者是被认为失败，该组将重新平衡以重新分配分区到另一个成员。

【讨论】：

我将 max.poll.interval.ms 更改为 Integer.MAX_VALUE 并且它起作用了。但想知道将其设置为更大的值有什么优点和缺点。
您可以确保只有当实际处理时间超过您的预期处理时间时，您的消费者才会被宣布死亡。该值应该是预期的最大处理时间。如果值太低，那么您的消费者被宣布过早死亡并且过于频繁地重新平衡。如果该值太高，那么消费者可能已经死亡，但在实际处理时间超过max.poll.interval.ms之前不会重新平衡。您必须微调属性以获得最佳性能。您可以进行基准测试，为您的消费者估算 WCET。

【解决方案2】：

造成这种情况的一个原因可能是您的 poll() 接收了很多消息，这就是为什么处理所有这些消息需要很长时间。
max.poll.records定义在对 poll() 的单个调用中返回的最大记录数。
根据kafka documentation，默认为500。
您可以尝试将其设置为更小的值，看看是否能解决您的问题。

【讨论】：

还可以考虑升级到最新的 Spring Kafka 和 Apache Kafka 本身：spring.io/projects/spring-kafka#learn。自从 Kafka 0.10.1: cwiki.apache.org/confluence/display/KAFKA/…
我同意...但是为什么我在其他实例听不同主题时没有遇到同样的问题
@ASHISHGUPTA 所以现在我们知道为什么其他实例没有看到相同的错误。作为下一步，您可能希望按照以下公式设置最大轮询间隔配置：max(300000,max(processing_time_of_first_instance, processing_time_of_second_instance))。这意味着将其设置为等于最长处理循环所花费的时间，但使其不小于默认值。
@ASHISHGUPTA 是的，但是一个实例提取的记录比其他实例少。您是否检查了每次调用 poll 是否在两个实例中返回相同数量的记录？
@ASHISHGUPTA 同样，当您知道单个记录的处理时间时，将其乘以 max.poll.records，这可以是您的 max.poll.interval.ms。

【解决方案3】：

在您的属性中设置 max.poll.interval.ms 最小值 5000000，例如

consumer.set("max.poll.interval.ms","5000000");
consumer.set("max.poll.records","2");
consumer.set("session.time.out.ms","30000"); 
consumer.set("heartbeat.interval.ms","25000");

希望：- 它会帮助你....

【讨论】：