kafka-consumer-api - 如果 group cordinator 长时间不调用 poll() 是否会认为 kafka 消费者 (0.9) 已死？

Question

https://www.safaribooksonline.com/library/view/kafka-the-definitive/9781491936153/ch04.html提到 “只要消费者定期发送心跳，就假定它还活着，并且正在处理来自其分区的消息。事实上，轮询消息的行为是导致消费者发送这些心跳的原因. 如果消费者停止发送心跳足够长的时间，它的 session 将超时，组协调器将认为它已经死亡并触发重新平衡。”

同样https://kafka.apache.org/090/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html指定 "broker 将使用心跳机制自动检测测试组中的失败进程。consumer 将定期自动 ping 集群，这让集群知道它是 alive。只要 consumer 能够这样做它被认为是活着的并且保留从分配给它的分区中使用的权利。如果它停止心跳的时间长于 session.timeout.ms 那么它将被认为是死的并且它的分区将被分配给另一个进程.”

在我的应用程序中，处理从先前的 poll() 接收的消息可能需要长达数小时，然后才会调用另一个 poll()。注意:我禁用了自动提交，因为我并不总是知道处理所有以前的消息需要多长时间。

a) 这会导致组协调器认为消费者已死亡或不活跃吗？

b) 是否有其他方法可以向组协调器发送心跳消息以保持 session 活跃？

c) session.timeout.ms 在这里对保持消费者活跃/活跃有什么影响吗？

Answer 1

a) 是的，如果你不调用 poll() 的时间超过 session.timeout.ms Kafka 认为消费者已经死了。

b) 作为替代方案，您可以在处理过程中调用 poll()(即与处理交错)以触发心跳(并在每个“真实”轮询之前查找)。使用额外的处理线程也是可能的，允许主线程定期轮询以发送心跳。 但是，您需要确保检测到处理线程上的故障(正确执行的技巧很棘手)!

c) 你可以增加超时值，但是，这可能不是你想要的，就好像你的消费者失败了，这个失败很晚才被检测到。

你描述的问题其实是已知的， future 消费者行为可能会改变。已经有关于它的讨论。参见 KIP-62了解更多详情。

更新

自 Kafka 0.10.1 以来，消费者有两个配置参数:max.poll.interval.ms 和 session.timeout.ms。第一个是两次连续轮询之间的最长时间，第二个是心跳超时。心跳在一个额外的线程中发送，因此现在与调用 poll() 分离。因此，增加 max.poll.interval.ms 不会产生无法快速检测到整个客户端故障(无心跳)的负面影响。