由于自动缩减/删除 pod，Kafka 中的重复消息消耗

Question

出身背景

我们有一个简单的生产者/消费者风格的应用程序，Kafka作为消息代理，消费者流程作为Kubernetes pods运行。我们定义了两个话题，即话题内话题和话题外话题。属于同一个消费者组的一组消费者pod从主题内读取消息，执行一些工作，并最终在工作完成后将相同的消息(密钥)写出到主题外。

问题描述

我们注意到，在 Kubernetes pod 中运行的消费者向外主题写出了重复的消息。换个说法，两个不同的使用者两次使用来自主题内消息的相同消息，因此也将相同的消息发布两次到主题外。我们分析了这个问题，可以肯定地得出结论，这个问题只有在 Kubernetes 自动缩小/删除 Pod 时才会发生。

事实上，我们有一个有趣的观察是，如果任何消息被两个不同的消费者从In主题中读取（因此在out主题中发布了两次），那么给定的消息始终是其中一个被缩小的pod使用的最后一条消息。换句话说，如果一条消息被消耗了两次，根本原因总是pod的缩小。

我们可以得出结论，在消费者将消息写入外主题之后，但在 Kafka 可以将偏移量提交到主题内之前，pod 正在缩小。

消费者配置

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "3600000");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG"org.apache.kafka.common.serialization.StringDeserializer")

动物园管理员/经纪人日志:

[2021-04-07 02:42:22,708] INFO [GroupCoordinator 0]: Preparing to rebalance group PortfolioEnrichmentGroup14 in state PreparingRebalance with old generation 1 (__consumer_offsets-17) (reason: removing member PortfolioEnrichmentConsumer13-9aa71765-2518-
493f-a312-6c1633225015 on heartbeat expiration) (kafka.coordinator.group.GroupCoordinator)
[2021-04-07 02:42:23,331] INFO [GroupCoordinator 0]: Stabilized group PortfolioEnrichmentGroup14 generation 2 (__consumer_offsets-17) (kafka.coordinator.group.GroupCoordinator)
[2021-04-07 02:42:23,335] INFO [GroupCoordinator 0]: Assignment received from leader for group PortfolioEnrichmentGroup14 for generation 2 (kafka.coordinator.group.GroupCoordinator)

我们尝试了什么

查看日志，很明显，由于检测信号过期而进行了重新平衡。我们添加了以下配置参数来增加检测信号并增加会话超时：

props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "10000")
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "900000");
props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "512");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");

然而，这并没有解决问题。查看代理日志，我们可以确认该问题是由于pod的缩减造成的。

问：当 Pod 缩小时，消息被消耗两次，这可能导致这种行为的原因是什么？

注：我已经了解问题的根本原因；然而，考虑到消费者是一个在无限循环中运行的长期进程，Kubernetes如何以及为什么在消费者提交偏移量之前缩减/杀死pod？我如何告诉Kubernetes在所有Kafka提交完成之前不要从用户组中删除正在运行的pod？

Answer 1

如果您希望消费者在退出前提交消息，则需要处理消费者的退出信号。很多语言都支持这一点。看看这个关于如何在java中实现这一点的线程-如何完成kafka消费者安全？（是否有在shutdownBook内调用线程#join的意思？）。

也就是说，请注意，没有100%的保证可以实现一次。甚至在给定时间运行任何退出清理之前，操作系统就可以强行终止您的进程(kill -9

Answer 2

“当pod缩小时，消息被消耗两次，是什么导致了这种行为？”

您自己已经提供了答案：“[…]在消费者将消息写入主题外但在Kafka可以将偏移量提交到主题内之前，pod正在缩小。”

由于消息已处理但未提交，因此在缩减发生后，另一个 Pod 将再次重新处理相同的消息。请记住，在使用者组中添加或删除使用者始终会启动重新平衡。您现在有了第一手经验，为什么通常应该尽可能避免这种情况。根据 Kafka 版本，重新平衡将导致消费者组的每个消费者停止消费，直到重新平衡完成。

为了解决您的问题，我看到了两个选项：

• 仅在正在运行的 Pod 处于空闲状态时将其从使用者组中删除
• 将使用者配置 auto.commit.interval.ms 减少到 1，因为这默认为 5 秒。仅当将 enable.auto.commit 设置为 true 时，此操作才有效。