1

背景

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

问题描述

我们注意到，在 Kubernetes pod 中运行的消费者正在向 out-topic 写入重复的消息。换句话说，两个不同的消费者从 in-topic 消费相同的消息两次，因此也两次向 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")

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

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

4

2 回答 2

1

“当 Pod 缩小规模时，消息被消费两次的原因可能是什么？”

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

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

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

  • 仅在空闲时将正在运行的 Pod 从 Consumer Group 中移除
  • 将消费者配置减少auto.commit.interval.ms到1默认为 5 秒。这仅在您设置 enable.auto.commit为时才有效true。
于 2021-04-12T12:49:44.837 回答
0

如果您希望您的消费者在退出之前提交消息，您将需要处理您的消费者的退出信号。很多语言都支持这一点。看看这个线程如何在 java 中执行此操作 -如何完成 kafka 消费者安全？（在 shutdownHook 中调用 thread#join 是否有意义？）。

话虽如此，请注意，不能 100% 保证只实现一次。甚至在给定时间运行任何退出清理（kill -9 <process_id >.

于 2021-05-26T15:38:29.313 回答