可能导致 Kafka 消息被重复利用的因素

有志者，事竟成！如果你在学习Golang，那么本文《可能导致 Kafka 消息被重复利用的因素》，就很适合你！文章讲解的知识点主要包括，若是你对本文感兴趣，或者是想搞懂其中某个知识点，就请你继续往下看吧~

昨天，我从日志中发现，在 kafka 组协调员发起组重新平衡后，kafka 正在重新消费一些消息。这些消息已在两天前被消耗（从日志中确认）。

日志中还报告了另外两个重新平衡，但它们不再重新使用消息。那么为什么第一次重新平衡会导致重新消费消息呢？存在哪些问题？

我正在使用golang kafka客户端。这是代码

config := sarama.newconfig()
config.version = version
config.consumer.offsets.initial = sarama.offsetoldest

并且我们在声明消息之前处理消息，所以我们似乎正在为 kafka 使用“至少发送一次”策略。我们在一台机器上有三个代理，在另一台机器上只有一个消费者线程（go 例程）。

对于这个现象有什么解释吗？ 我认为这些消息一定已经提交了，因为它们是在两天前被消耗的，或者为什么 kafka 会在没有提交的情况下保留偏移量超过两天？

使用代码示例：

func (consumer *Consumer) ConsumeClaim(session 
sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {

for message := range claim.Messages() {
    realHanlder(message)   // consumed data here
    session.MarkMessage(message, "") // mark offset
}

return nil
}

添加：

1. 应用程序重新启动后发生重新平衡。还有另外两次重新启动并没有导致重新启动

2. kafka的配置

   log.retention.check.interval.ms=300000

   log.retention.hours=168

   zookeeper.connection.timeout.ms=6000

   group.initial.rebalance.delay.ms=0

   删除.topic.enable = true

   auto.create.topics.enable=false

解决方案

通过阅读golang saram客户端和kafka服务器的源码，终于找到原因如下

1. 消费者组偏移量保留时间为24小时，这是kafka的默认设置，而日志保留时间为我们明确规定的 7 天。

2. 我的服务器应用程序运行在测试环境中，很少有人可以访问，这意味着kafka生产者产生的消息可能很少，然后消费者组没有多少消息可以消费，因此消费者可能不会提交任何消息长时间偏移。

3. 当消费偏移量超过24小时没有更新时，由于偏移量配置，kafka代理/协调器将从分区中删除消费偏移量。下次 saram 从 kafka 代理查询偏移量在哪里时，客户端当然什么也得不到。请注意，我们使用 sarama.OffsetOldest 作为初始值，然后 sarama 客户端将从 kafka 代理保留的消息开头开始消费消息，这会导致消息重新消费，而这种情况很可能会发生，因为日志保留是7天

到这里，我们也就讲完了《可能导致 Kafka 消息被重复利用的因素》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！