昨天，我從日志中發(fā)現(xiàn)，在 Kafka 組協(xié)調(diào)員發(fā)起組重新平衡后，kafka 正在重新消費(fèi)一些消息。這些消息已在兩天前被消耗（從日志中確認(rèn)）。日志中還報(bào)告了另外兩個(gè)重新平衡，但它們不再重新使用消息。那么為什么第一次重新平衡會(huì)導(dǎo)致重新消費(fèi)消息呢？存在哪些問題？我正在使用 golang kafka 客戶端。這是代碼config := sarama.NewConfig()
config.Version = version
config.Consumer.Offsets.Initial = sarama.OffsetOldest并且我們?cè)诼暶飨⒅疤幚硐ⅲ晕覀兯坪跽跒?kafka 使用“至少發(fā)送一次”策略。我們?cè)谝慌_(tái)機(jī)器上有三個(gè)代理，在另一臺(tái)機(jī)器上只有一個(gè)消費(fèi)者線程（go 例程）。對(duì)于這種現(xiàn)象有什么解釋嗎？我認(rèn)為這些消息肯定已經(jīng)提交了，因?yàn)樗鼈兪窃趦商烨氨幌牡模蛘邽槭裁?kafka 會(huì)在沒有提交的情況下保留偏移量超過兩天？消費(fèi)代碼示例：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}添加：應(yīng)用程序重新啟動(dòng)后發(fā)生重新平衡。還有另外兩次重新啟動(dòng)并沒有導(dǎo)致重新啟動(dòng)卡夫卡的配置log.retention.check.interval.ms=300000log.retention.hours=168zookeeper.connection.timeout.ms=6000group.initial.rebalance.delay.ms=0delete.topic.enable = trueauto.create.topics .enable=假
1 回答
ITMISS
TA貢獻(xiàn)1871條經(jīng)驗(yàn) 獲得超8個(gè)贊
通過閱讀golang saram客戶端和kafka服務(wù)器的源碼，最終找到原因如下
消費(fèi)者組偏移保留時(shí)間為24 小時(shí)，這是 kafka 默認(rèn)設(shè)置，而日志保留時(shí)間是我們明確設(shè)置的 7 天。
我的服務(wù)器應(yīng)用程序運(yùn)行在測(cè)試環(huán)境中，很少有人可以訪問，這意味著kafka生產(chǎn)者產(chǎn)生的消息可能很少，然后消費(fèi)者組也沒有多少消息可以消費(fèi)，因此消費(fèi)者可能很長時(shí)間不會(huì)提交任何偏移量。
當(dāng)消費(fèi)偏移量超過24小時(shí)未更新時(shí)，由于偏移量配置，kafka代理/協(xié)調(diào)器將從分區(qū)中刪除消費(fèi)偏移量。下次 saram 從 kafka 代理查詢偏移量在哪里時(shí)，客戶端當(dāng)然什么也得不到。請(qǐng)注意，我們使用sarama.OffsetOldest作為初始值，然后 sarama 客戶端將從 kafka Broker 保留的消息開頭開始消費(fèi)消息，這會(huì)導(dǎo)致消息重新消費(fèi)，并且這種情況很可能發(fā)生，因?yàn)槿罩颈Ａ羝跒?7天
- 1 回答
- 0 關(guān)注
- 152 瀏覽
添加回答
舉報(bào)
0/150
提交
取消