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

如何解决由于自动缩小/删除 Pod，Kafka 中的重复消息消耗

背景

我们有一个简单的生产者/消费者风格的应用程序，其中 Kafka 作为消息代理，消费者进程作为 Kubernetes pod 运行。我们定义了两个主题，即主题内和主题外。属于同一个消费者组的一组消费者 Pod 从 in-topic 读取消息，执行一些工作，最后在工作完成后将相同的消息（key）写出到 out-topic。

问题描述

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

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

我们可以得出结论，在消费者将消息写入 out-topic 之后，但在 Kafka 可以将偏移量提交到 in-topic 之前，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")

Zookeeper/broker 日志：

[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？

解决方法

“当 pod 缩小规模时，一条消息会被消费两次，这可能是什么原因造成的？”

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

>

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

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

• 仅在空闲时将正在运行的 Pod 从消费者组中移除
• 将使用者配置 auto.commit.interval.ms 减少到 1，因为这默认为 5 秒。这仅在您将 enable.auto.commit 设置为 true 时有效。

,

如果您希望您的消费者在退出之前提交 message/s，您需要处理给您的消费者的退出信号。许多语言确实支持这一点。看看这个线程如何在 java - How to finish kafka consumer safety?(Is there meaning to call thread#join inside shutdownHook ? ) 中做到这一点。

话虽如此，请注意，没有 100% 保证只实现一次。您的进程甚至可以在给定时间运行任何退出清理之前被操作系统强制终止 (kill -9 >.