当消费者缓慢时，Kafka 持续重新平衡

Question

我一直在尝试测试我们的 Kafka 是否有一些负面情况，其中一个是非常慢的消费者。我在我的@KafkaListener 方法中设置了Thread.sleep(15000)（它是spring-kafka）并将并发设置为3。我有1 个主题和1 个分区。 我将 10 条消息放入主题并启动服务。 当3个消费者开始时，他们都到达(Re-)joining group点， 但是只有其中一个（假设它是consumer-2）会到达：

Successfully joined group with generation X

然后开始慢慢消费消息。

（顺便说一句，我使用 MANUAL_IMMEDIATE Ack 模式，但即使我不向侦听器添加 Acknowledgement 参数并且不确认消息，它也是可重现的）。 我接下来看到的内容如下： 直到消费者 2 处理所有消息，每 3 秒（默认心跳间隔）我在控制台中收到一条消息：

AbstractCoordinator$HeartbeatResponseHandler: [Consumer clientId=consumer-2, groupId=pixel-group] Attempt to heartbeat failed since group is rebalancing

我想知道为什么会这样。只有在所有 10 条消息都是进程之后，才会有另一个重新平衡，之后所有 3 个消费者都将打印：

Successfully joined group with generation X

其中一个将被分配一个分区，不再有心跳问题。 这只发生在我将睡眠间隔设置为高于心跳间隔的值时。它通常在所有消费者都启动时发生一次，但很快就会成功设置。

所以，总结起来似乎是：

如果消费者处理时间> 心跳间隔时间 - 除了第一个消费者之外的所有消费者都无法完成重新平衡（他们可能无法与他们缓慢的领导者交谈）。 我不明白为什么这个心跳错误如此持久？ 如果睡眠时间比心跳时间长，为什么其他消费者不能在领导者的消息消费之间完成重新平衡？

更新 卡夫卡版本 2.12-2.2.0 Spring-Kafka 2.2.3.RELEASE

Answer 1

...并发到 3。我有 1 个主题和 1 个分区...

您至少需要与消费者一样多的分区 - 一个分区只能由一个消费者使用。

你用的是什么版本？从 KIP-62 (Kafka 0.10.1.0) 开始，心跳由 kafka-clients 在后台发送。因此，仅当侦听器花费的时间超过max.poll.interval.ms 时，才应进行重新平衡。谷歌 KIP-62 了解更多信息。

编辑

当你的听众正在睡觉时，你应该会看到这样的日志......

2019-06-13 09:47:52.008 DEBUG 61914 --- [hread | rbgh664] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=source, groupId=rbgh664] Heartbeat thread started
...
2019-06-13 09:47:52.072  INFO 61914 --- [           main] com.example.Rbgh664Application           : Sleeping for 15
2019-06-13 09:47:55.120 DEBUG 61914 --- [hread | rbgh664] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=source, groupId=rbgh664] Sending Heartbeat request to coordinator localhost:9092 (id: 2147483647 rack: null)
2019-06-13 09:47:55.121 TRACE 61914 --- [hread | rbgh664] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=source, groupId=rbgh664] Sending HEARTBEAT {group_id=rbgh664,generation_id=45,member_id=source-82297cee-063a-4e0d-89d0-15cfbd6ef680} with correlation id 10 to node 2147483647
2019-06-13 09:47:55.226 TRACE 61914 --- [hread | rbgh664] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=source, groupId=rbgh664] Completed receive from node 2147483647 for HEARTBEAT with correlation id 10, received {throttle_time_ms=0,error_code=0}
2019-06-13 09:47:55.227 DEBUG 61914 --- [hread | rbgh664] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=source, groupId=rbgh664] Received successful Heartbeat response
2019-06-13 09:47:58.120 DEBUG 61914 --- [hread | rbgh664] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=source, groupId=rbgh664] Sending Heartbeat request to coordinator localhost:9092 (id: 2147483647 rack: null)
2019-06-13 09:47:58.120 TRACE 61914 --- [hread | rbgh664] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=source, groupId=rbgh664] Sending HEARTBEAT {group_id=rbgh664,generation_id=45,member_id=source-82297cee-063a-4e0d-89d0-15cfbd6ef680} with correlation id 11 to node 2147483647
2019-06-13 09:47:58.225 TRACE 61914 --- [hread | rbgh664] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=source, groupId=rbgh664] Completed receive from node 2147483647 for HEARTBEAT with correlation id 11, received {throttle_time_ms=0,error_code=0}
2019-06-13 09:47:58.226 DEBUG 61914 --- [hread | rbgh664] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=source, groupId=rbgh664] Received successful Heartbeat response
2019-06-13 09:48:01.203 DEBUG 61914 --- [hread | rbgh664] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=source, groupId=rbgh664] Sending Heartbeat request to coordinator localhost:9092 (id: 2147483647 rack: null)
2019-06-13 09:48:01.204 TRACE 61914 --- [hread | rbgh664] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=source, groupId=rbgh664] Sending HEARTBEAT {group_id=rbgh664,generation_id=45,member_id=source-82297cee-063a-4e0d-89d0-15cfbd6ef680} with correlation id 12 to node 2147483647
2019-06-13 09:48:01.310 TRACE 61914 --- [hread | rbgh664] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=source, groupId=rbgh664] Completed receive from node 2147483647 for HEARTBEAT with correlation id 12, received {throttle_time_ms=0,error_code=0}
2019-06-13 09:48:01.310 DEBUG 61914 --- [hread | rbgh664] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=source, groupId=rbgh664] Received successful Heartbeat response
2019-06-13 09:48:04.285 DEBUG 61914 --- [hread | rbgh664] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=source, groupId=rbgh664] Sending Heartbeat request to coordinator localhost:9092 (id: 2147483647 rack: null)
2019-06-13 09:48:04.286 TRACE 61914 --- [hread | rbgh664] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=source, groupId=rbgh664] Sending HEARTBEAT {group_id=rbgh664,generation_id=45,member_id=source-82297cee-063a-4e0d-89d0-15cfbd6ef680} with correlation id 13 to node 2147483647
2019-06-13 09:48:04.390 TRACE 61914 --- [hread | rbgh664] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=source, groupId=rbgh664] Completed receive from node 2147483647 for HEARTBEAT with correlation id 13, received {throttle_time_ms=0,error_code=0}
2019-06-13 09:48:04.390 DEBUG 61914 --- [hread | rbgh664] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=source, groupId=rbgh664] Received successful Heartbeat response
20
...