go - 重复消费 Kafka 消息的可能原因
问题描述
昨天我从日志中发现,在 Kafka 组协调器启动组重新平衡后,kafka 正在重新消费一些消息。这些消息已在两天前被消费(从日志中确认)。
日志中报告了另外两个重新平衡,但它们不再重新使用消息。那么为什么第一次重新平衡会导致重复消费消息呢?有什么问题?
我正在使用 golang kafka 客户端。这是代码
config := sarama.NewConfig()
config.Version = version
config.Consumer.Offsets.Initial = sarama.OffsetOldest
我们在声明消息之前处理消息,所以似乎我们正在使用 kafka 的“至少发送一次”策略。我们在一台机器上有三个代理,而在另一台机器上只有一个消费者线程(goroutine)。
对此现象有何解释?我认为这些消息一定已经提交了,因为它们在两天前被消费了,或者为什么 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
}
添加:
应用程序重新启动后发生重新平衡。还有另外两次重启并没有导致重新消费
卡夫卡的配置
log.retention.check.interval.ms=300000
log.retention.hours=168
zookeeper.connection.timeout.ms=6000 group.initial.rebalance.delay.ms
=0
delete.topic.enable = true
auto.create.topics .enable=false
解决方案
通过阅读golang saram客户端和kafka服务器的源码,终于找到了原因如下
消费者组偏移保留时间是24 小时,这是 kafka 的默认设置,而日志保留时间是我们明确设置的 7 天。
我的服务器应用运行在测试环境,很少人可以访问,这意味着kafka生产者产生的消息可能很少,然后消费者组消费的消息很少,因此消费者可能很长时间不会提交任何偏移量。
当消费偏移量超过 24 小时未更新时,由于偏移量配置,kafka 代理/协调器将从分区中删除消费偏移量。下次 saram 从 kafka 代理查询偏移量在哪里,当然客户端什么也得不到。注意我们使用sarama.OffsetOldest作为初始值,然后 sarama 客户端会从 kafka broker 保存的消息开始消费消息,这会导致消息重复消费,这很可能发生,因为日志保留是7 天
推荐阅读
- c# - 证书安装 (Linux)
- f# - 从 Visual Studio 中查找所有引用被 F# 成员破坏
- javascript - 居中对齐图例文本和圆圈
- webpack - 包含外部 ES6 库时,ExtJs 测试/生产构建失败
- symfony5 - 找不到路线
- django-views - django 形式的馈送数据未传递到数据库或从数据库中检索
- javascript - 数据绑定返回带有数据绑定的 html a href
- flutter - 我正在使用 showGeneralDialog 我需要根据对话框内的布尔值重新渲染按钮
- ios - CoreTelephony 在多平台项目中的使用
- r - 正则表达式:如何将 .* 限制为仅数字和标点符号