[[438994]]

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

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

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

這樣嚴苛的設置，那為什么還會出現消息丟失呢?請聽筆者慢慢道來。

1、故障現象

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

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

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

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

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

2、問題分析

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

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

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

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

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

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

消費組重平衡：當主題分區數量或消費者數量發生變化后，消費者之間需要對分區進行重新分配，實現消費端端負載均衡。

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

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

無外乎如下兩個原因：

• 服務端丟失位點，導致客戶端無法查詢到位點
• 客戶端主動向服務端提交了-1，導致位點丟失

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

查看客戶端提交位點的API，發現用于封裝客戶端位點的實體類會對位點進行校驗，代碼截圖如下：

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

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

2.1 客戶端位點查找機制

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

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

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

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

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

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

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

• Setting offset for partition 第一次查詢時找到了位點，并且不為-1，也不是最早位點。
• Found no committed offset for partition 后面反復進行重平衡，反復查詢日志，竟然后面無法正確查詢到位點，而是返回沒有找到位點(返回-1)。
• Resetting offset for partition XX to offset XXXX. 根據重置策略進行了位點重置。

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

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

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

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

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

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

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

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

本文就說到這里了，關于Kafka集群為什么會出現大量__consumer_offsets進行Leader選舉，后續文章會一一展開，敬請持續關注我。

3、感想

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

正例如大部分朋友應該知道kafka在后續版本中的消費位點是存儲在系統主題__consumer_offsets中，但又有多少人知道，這個主題的分區一旦出現Leader選舉，伴隨而來的是一大堆消費組全部發生重平衡，導致消費組停止消費呢?

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

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