[[438994]]

講真，我今年的雙十一有點“背”，負(fù)責(zé)的Kafka集群出了一些幺蛾子，但正是這些幺蛾子，讓我這個雙十一過的非常充實，也讓我意識到如果不體系化學(xué)習(xí)Kafka，是無法做到生產(chǎn)集群及時預(yù)警，將故障扼殺在搖籃中，因此也下定決心研讀Kafka的內(nèi)核。

本文就先來分享一個讓我始料未及的故障：Kafka生產(chǎn)環(huán)境大面積丟失消息。

首先要闡述的是消息丟失并不是因為斷電，而且集群的副本數(shù)量為3，消息發(fā)送端設(shè)置的acks=-1(all)。

這樣嚴(yán)苛的設(shè)置，那為什么還會出現(xiàn)消息丟失呢?請聽筆者慢慢道來。

1、故障現(xiàn)象

故障發(fā)生時，接到多個項目組反饋說消費組的位點被重置到幾天前了，截圖如下：

從上面的消費組延遲監(jiān)控曲線上來看，一瞬間積壓數(shù)從零直接飆升，初步懷疑是位點被重置了。

那位點為什么會被重置呢?

什么?你這篇文章不是說要講Kafka為什么會丟消息嗎?怎么你又扯說消費組位點被重置呢?標(biāo)題黨!!!

NO、NO、NO，各位看官，絕對不是文不對題，請帶著這個疑問，與我共同探究吧。

2、問題分析

遇到問題，莫慌，講道理，基于MQ的應(yīng)用，消費端一般都會實現(xiàn)冪等，也就是消息可以重復(fù)被處理，并且不會影響業(yè)務(wù)，故解決的方式就是請項目組先評估一下，先人工將位點設(shè)置到出現(xiàn)問題的前30分鐘左右，快速止血。

一波操作猛如虎，接下來就得好好分析問題產(chǎn)生的原因。

通過查看當(dāng)時Kafka服務(wù)端的日志(server.log)，可以看到如下日志：

上面的日志被修改的“面目全非”，其關(guān)鍵日志如下：

• Member consumer-1-XX in group consumerGroupName has failed, removing it from the group
• Preparing to rebalance group XXXX on heartbeat expiration

上面的日志指向性非常明顯：由于心跳檢測過期，消費組協(xié)調(diào)器將消費者從消費組中移除，重而觸發(fā)重平衡。

消費組重平衡：當(dāng)主題分區(qū)數(shù)量或消費者數(shù)量發(fā)生變化后，消費者之間需要對分區(qū)進(jìn)行重新分配，實現(xiàn)消費端端負(fù)載均衡。

消息消費者在重平衡期間消費會全部暫停，當(dāng)消費者重新完成分區(qū)的負(fù)載均衡后，繼續(xù)從服務(wù)端拉起消息，此時消費端并不知道從哪個位置開始，故需要從服務(wù)端查詢位點，使得消費者能從上次消費的位點繼續(xù)消費。

現(xiàn)在出現(xiàn)消費位點被重置到最早位點，可以理解為位點丟失?那為什么會丟失位點呢?

無外乎如下兩個原因：

• 服務(wù)端丟失位點，導(dǎo)致客戶端無法查詢到位點
• 客戶端主動向服務(wù)端提交了-1，導(dǎo)致位點丟失

目前我們公司使用的Kafka版本為2.2.x，消費組的位點是存儲在一個系統(tǒng)主題(__consumer_offsets)中，無論是服務(wù)器級別還是Topic級別，參數(shù)unclean.leader.election.enable都是設(shè)置為false，表示只有ISR集合中的副本才能參與Leader選舉，這樣就能嚴(yán)格保證位點消息并不會丟失或回到歷史某一個位點。

查看客戶端提交位點的API，發(fā)現(xiàn)用于封裝客戶端位點的實體類會對位點進(jìn)行校驗，代碼截圖如下：

如果傳入的位點為-1，直接會拋出異常，故客戶端并沒有機(jī)會向服務(wù)端提交-1的位點，那位點為什么會丟失呢?

為了進(jìn)一步探究，我們不得不將目光投向消費組在初次時是如何獲取位點，從源碼的角度去分析，從而尋找關(guān)鍵日志，并對日志文件進(jìn)行對照，嘗試得到問題的解。

2.1 客戶端位點查找機(jī)制

為了探究客戶端的位點獲取機(jī)制，筆者詳細(xì)閱讀了消費者在啟動時的流程，具體入口為KafkaConsumer的poll方法，其詳細(xì)流程圖如下所示：

上述的核心要點說明如下：

• 在消費者(KafkaConsumer)的poll方法消息時會調(diào)用updateAssignmentMetadataIfNeeded方法，該方法主要執(zhí)行消費組初始化、消費組重平衡、獲取消費位點等與元數(shù)據(jù)相關(guān)工作。
• 如果當(dāng)前消費組訂閱的分區(qū)(重平衡后分配的分區(qū))都存在位點，則返回true，說明無需更新位點。
• 如果當(dāng)前存在分配的分區(qū)沒有正確的位點(例如一次重平衡后新增加的分區(qū))，此時需要向服務(wù)端發(fā)送查找位點請求，服務(wù)端查詢__consumer_offsets主題，返回位點信息。
• 如果查詢到位點，輸出DEBUG級別日志(Setting offset for partition),輸出從服務(wù)端查詢到的位點;如果未查詢到位點，同樣會輸出DEBUG級別日志(Found no committed offset for partition)。
• 如果沒有查詢到位點，則需要根據(jù)消費組配置的位點重置策略，其具體配置參數(shù)：auto.offset.reset,其可選值：
  • latest 最新位點
  • earliest 最早位點
  • none 不重置位點
• 如果重置位點選擇的是none，則會拋出NoOffsetForPartitionException異常。
• 如果重置位點選擇的是latest、earliest，則消費者將從查詢到的位點開始消費，并輸出DEBUG級別日志(Resetting offset for partition XX to offset XXXX.)
• 非常遺憾，消費者的位點查找機(jī)制，Kafka客戶端打印的過程日志是DEBUG級別，這在生產(chǎn)環(huán)境基本是不會輸出的，給我排查問題(找到足夠的證據(jù))帶來了不便。

這里不得不吐槽一下Kafka輸出日志的策略：位點的變更是一個非常關(guān)鍵的狀態(tài)變更，而且輸出這些日志的頻率不會很大，日志級別應(yīng)該使用INFO，而不是DEBUG。

Kafka的日志是Debug，故當(dāng)時是無法找到證據(jù)進(jìn)行輔助說明，只能排查出為什么會因為心跳超時而觸發(fā)重平衡。

溫馨提示：關(guān)于心跳為什么會超時，從而觸發(fā)重平衡原因，將會在后續(xù)的故障分析相關(guān)的文章中詳細(xì)闡述。

找到重平衡觸發(fā)原因后，在測試環(huán)境進(jìn)行壓測并加以重現(xiàn)，同時將客戶端日志級別設(shè)置為debug，從而查找證據(jù)，功夫不負(fù)有心人，完美的找到了上文中提到的三條日志：

• Setting offset for partition 第一次查詢時找到了位點，并且不為-1，也不是最早位點。
• Found no committed offset for partition 后面反復(fù)進(jìn)行重平衡，反復(fù)查詢?nèi)罩荆谷缓竺鏌o法正確查詢到位點，而是返回沒有找到位點(返回-1)。
• Resetting offset for partition XX to offset XXXX. 根據(jù)重置策略進(jìn)行了位點重置。

從上面的日志分析，也可以明確地出結(jié)論，服務(wù)端是有存儲消費組的位點的，不然不會出現(xiàn)第一條日志，成功找到了一個有效的位點，只是在后續(xù)重平衡過程中，多次需要查詢位點時，反而返回了-1，那服務(wù)端在什么情況下返回-1呢?

Broker服務(wù)端處理心跳包的入口是kafkaApis的handleOffsetFetchRequest方法，找到獲取位點的關(guān)鍵代碼，如下所示：

從上面來看，服務(wù)端返回INVALID_OFFSET = -1L的情況如下：

• 消費組元信息管理器中的緩存(內(nèi)存)中并不存在該消費組，將返回-1，那又在什么情況下服務(wù)端會沒有正在使用的消費組元信息呢?
  • __consumer_offsets主題的分區(qū)發(fā)生Leader選舉，當(dāng)前Broker中擁有的分區(qū)變更為follower后，與該分區(qū)對應(yīng)的消費組的元信息將被移除。為什么會這樣呢?這里背后的原因是Kafka中的消費組在Broker端需要選舉出一個組協(xié)調(diào)器，用于協(xié)調(diào)消費組的重平衡，選舉算法就是將消費組的名稱取hashcode，得到的值與 consumer_offsets主題的分區(qū)數(shù)取模得到一個分區(qū)數(shù)，然后該分區(qū)的Leader節(jié)點所在的Broker為該消費組的組協(xié)調(diào)器，故分區(qū)Leader發(fā)生變化，與之關(guān)聯(lián)的消費組的組協(xié)調(diào)器需要重新選舉。
  • 刪除消費組時將器移出。
• 消費組的狀態(tài)為GroupState.Dead 消費組狀態(tài)變更為Dead，通常有如下幾種情況：
  • 消費組被刪除
  • __consumer_offsets分區(qū)leader發(fā)生變化，觸發(fā)位點重新加載，要先將消費組狀態(tài)變更為Dead，然后新的分區(qū)Leader所在機(jī)器上會加載新的位點，然后引導(dǎo)消費組重平衡。

服務(wù)端中并沒有存儲該消費組的位點信息，說明該消費組還未提交過位點

那上面的情況，對于一個正在運行許久的消費組來說，上述這些情況會發(fā)生嗎?查找服務(wù)端相關(guān)日志，可以明確看到大量__consumer_offsets相關(guān)分區(qū)發(fā)生leader選舉，容易觸發(fā)上述第一種情況，這樣消費組發(fā)起的Offset Fetch請求是有可能返回-1，從而會引導(dǎo)消費組根據(jù)重置策略進(jìn)行位點重置。

查看文章開頭部分，消費組設(shè)置的重置策略選的是earliest，消費組在一瞬間消費積壓從0飆升到幾個億，就能解釋的通了。

看到這里，大家是不是會突然“后背發(fā)涼”，如果消費組配置的位點重置策略(auto.offset.reset)為latest，是不是很容易引起消息丟失，即一部分消費被跳過而不被消費，示意圖說明如下：

本文就說到這里了，關(guān)于Kafka集群為什么會出現(xiàn)大量__consumer_offsets進(jìn)行Leader選舉，后續(xù)文章會一一展開，敬請持續(xù)關(guān)注我。

3、感想

講真，由于Kafka服務(wù)端使用的編程語言為scala，筆者并沒有嘗試去看Kafka的源碼，只是詳細(xì)剖析了Kafka的消息發(fā)送、消息消費機(jī)制，本以為可以輕松駕馭公司各個項目關(guān)于Kafka使用層面的問題，但事實上也是如此，對項目組的咨詢我應(yīng)對起來得心應(yīng)手，但一旦服務(wù)端出現(xiàn)問題，還是會有點茫然，當(dāng)然我們有一套完備的集群問題出現(xiàn)應(yīng)急方案，但一旦出現(xiàn)問題，盡管你能快速恢復(fù)，但故障一旦發(fā)生，損失就無法避免，故我們還是要對自己負(fù)責(zé)的內(nèi)容研究透，提前做好巡檢、根據(jù)體系化的知識提前規(guī)避故障的發(fā)生。

正例如大部分朋友應(yīng)該知道kafka在后續(xù)版本中的消費位點是存儲在系統(tǒng)主題__consumer_offsets中，但又有多少人知道，這個主題的分區(qū)一旦出現(xiàn)Leader選舉，伴隨而來的是一大堆消費組全部發(fā)生重平衡，導(dǎo)致消費組停止消費呢?

故筆者將下定決心，好好閱讀一下kafka服務(wù)端相關(guān)源碼，成體系化理解Kafka，在工作中更好的駕馭Kafka，《Kafka原理與實戰(zhàn)》專欄在路上，有興趣的朋友可以點擊文章前的標(biāo)簽加以關(guān)注。 

最后，期待您的點贊，您的點贊也是我最大的動力，我們下回見。