1、背景

最近在折騰Kafka日志集群，由于公司部署的應(yīng)用不斷增加，日志采集程序?qū)⒉杉降娜罩景l(fā)送到Kafka集群時出現(xiàn)了較大延遲，總的TPS始終上不去，為了不影響業(yè)務(wù)團(tuán)隊(duì)通過日志排查問題，采取了先解決問題，再排查的做法，對Kafka集群進(jìn)行擴(kuò)容，但擴(kuò)容后尷尬的是新增加的5臺機(jī)器中，有兩臺機(jī)器的消費(fèi)發(fā)送響應(yīng)時間比其他機(jī)器明顯高出不少，為了確保消息服務(wù)的穩(wěn)定性，又臨時對集群進(jìn)行縮容，將這臺機(jī)器從集群中剔除，具體的操作就是簡單粗暴的使用 kill pid命令，但意外發(fā)生了。

發(fā)現(xiàn)Java客戶端報如下錯誤：

而Go客戶端報的錯誤如下所示：

基本可以認(rèn)為是部分分區(qū)沒有在線Leader,無法成功發(fā)送消息。

2、問題分析

那為什么會出現(xiàn)這個問題嗎？Kafka一個節(jié)點(diǎn)下線，不是會自動觸發(fā)故障轉(zhuǎn)移，分區(qū)leader不是會被重新選舉嗎？請帶著這個疑問，開始我們今天的探究之旅。

首先我們可以先看看當(dāng)前存在問題的分區(qū)的路由信息，從第一張圖中看出主題dw_test_kafka_0816000的101分區(qū)消息發(fā)送失敗，我們在Zookeeper中看一下其狀態(tài)，具體命令如下：

./zkCli.sh -server 127.0.0.1:2181
get -s  /kafka_cluster_01/brokers/topics/dw_test_kafka_0816000/partitions/101/state

該命令可以看到對應(yīng)分區(qū)的相信信息，如下圖所示：

這里顯示出leader的狀態(tài)為-1,而isr列表中只有一副本，在broker-1上，但此時broker id為1的機(jī)器已經(jīng)下線了，那為什么不會觸發(fā)分區(qū)Leader重新選舉呢？

其實(shí)看到這里，我相信你只要稍微細(xì)想一下，就能發(fā)現(xiàn)端倪，isr字段的值為1,說明該分區(qū)的副本數(shù)為1，說明該分區(qū)只在一個Broker上存儲數(shù)據(jù)，一旦Broker下線，由于集群內(nèi)其他Broker上并沒有該分區(qū)的數(shù)據(jù)，此時是無法進(jìn)行故障轉(zhuǎn)移的，因?yàn)橐坏┮M(jìn)行故障轉(zhuǎn)移，分區(qū)的數(shù)據(jù)就會丟失，這樣帶來的影響將是非常嚴(yán)重的。

那為什么該主題的副本數(shù)會設(shè)置為1呢？那是因?yàn)楫?dāng)時集群的壓力太大，節(jié)點(diǎn)之間復(fù)制數(shù)據(jù)量巨大，網(wǎng)卡基本滿負(fù)荷在運(yùn)轉(zhuǎn)，而又是日志集群，對數(shù)據(jù)的丟失的接受程度較大，故當(dāng)時為了避免數(shù)據(jù)在集群之間的大量復(fù)制，將該主題的副本數(shù)設(shè)置為了1。

但集群節(jié)點(diǎn)的停機(jī)維護(hù)是少不了的，總不能每一次停機(jī)維護(hù)，都會出現(xiàn)一段時間數(shù)據(jù)寫入失敗吧。要解決這個問題，我們在停機(jī)之前，需要先對主題進(jìn)行分區(qū)移動，將該主題的分區(qū)從需要停機(jī)的集群中移除。

主題分區(qū)移動的具體做法，請參考我之前的一篇文章Kafka主題遷移實(shí)踐 的第三部分。

3、Kafka節(jié)點(diǎn)下線分區(qū)的故障轉(zhuǎn)移機(jī)制

Kafka單副本的主題在集群內(nèi)一臺節(jié)點(diǎn)下線后，將無法完成分區(qū)的故障轉(zhuǎn)移機(jī)制，為了深入掌握底層的一些實(shí)現(xiàn)細(xì)節(jié)，我想再深入探究一下kafka節(jié)點(diǎn)下線的一些故障轉(zhuǎn)移機(jī)制。

溫馨提示：接下來主要是從源碼角度深入探究實(shí)現(xiàn)原理，加深對這個過程的理解，如果大家不感興趣，可以直接進(jìn)入到本文的第4個部分：總結(jié)。

在Kafka中依賴的Zookeeper服務(wù)器上存儲了當(dāng)前集群內(nèi)存活的broker信息，具體的路徑為/{namespace}/brokers/brokers/ids，具體圖示如下：

并且ids下的每一個節(jié)點(diǎn)記錄了Broker的一些信息，例如對外提供服務(wù)的協(xié)議、端口等，值得注意的是這些節(jié)點(diǎn)為臨時節(jié)點(diǎn)，如下圖所示：

這樣一旦對應(yīng)的Broker宕機(jī)下線，對應(yīng)的節(jié)點(diǎn)會刪除，Kafka集群內(nèi)的Controller角色在啟動時會監(jiān)聽該節(jié)點(diǎn)下節(jié)點(diǎn)的變化，并作出響應(yīng)，最終將會調(diào)用KafkaController的onBrokerFailure方法,具體代碼如下所示：

這個方法實(shí)現(xiàn)比較復(fù)雜，我們在這里不做過多分散，重點(diǎn)查找分區(qū)的故障轉(zhuǎn)移機(jī)制，也就是接下來我們將具體分析KafkaController的onReplicasBecomeOffline方法，主要探究分區(qū)的故障轉(zhuǎn)移機(jī)制。

3.1 onReplicasBecomeOffline故障轉(zhuǎn)移

由于該方法實(shí)現(xiàn)復(fù)雜，接下來將分布對其進(jìn)行詳解。

Step1：從需要設(shè)置為下線狀態(tài)分區(qū)進(jìn)行分組，分組依據(jù)為是否需要刪除，沒有觸發(fā)刪除的集合用newofflineReplicasNotForDeletion表示，需要被刪除的集合用newofflineReplicasForDeletion表示。

Step2：挑選沒有Leader的分區(qū)，用partitionsWithoutLeader，代碼如下圖所示：

分區(qū)沒有Leader的標(biāo)準(zhǔn)是：分區(qū)的Leader副本所在的Broker沒有下線，并且沒有被刪除。

Step3：將沒有Leader的分區(qū)狀態(tài)變更為OfflinePartition（離線狀態(tài)）,這里的狀態(tài)更新是放在kafka Controller中的內(nèi)存中，具體的內(nèi)存結(jié)構(gòu)：Map[TopicPartition, PartitionState]。

Step4：Kafka分區(qū)狀態(tài)機(jī)驅(qū)動(觸發(fā)）分區(qū)狀態(tài)為OfflinePartition、NewPartition向OnlinePartition轉(zhuǎn)化，狀態(tài)的轉(zhuǎn)化主要包括兩個重要的步驟：

調(diào)用PartitionStateMachine的doHandleStateChanges的方法，驅(qū)動分區(qū)狀態(tài)機(jī)的轉(zhuǎn)換。

然后調(diào)用ControllerBrokerRequestBatch的sendRequestsToBrokers方法，實(shí)現(xiàn)元信息在其他Broker上的同步。

由于篇幅的問題，我們這篇文章不會體系化的介紹Kafka分區(qū)狀態(tài)機(jī)的實(shí)現(xiàn)細(xì)節(jié)，先重點(diǎn)關(guān)注OfflinePartition離線狀態(tài)向OnlinePartition轉(zhuǎn)化過程。

我們首先說明一下OfflinePartition離線狀態(tài)向OnlinePartition轉(zhuǎn)化過程時各個參數(shù)的含義：

Seq[TopicPartition] partitions 當(dāng)前處于OfflinePartition、NewPartition狀態(tài)、并且沒有刪除的分區(qū)。

PartitionState targetState 狀態(tài)驅(qū)動的目標(biāo)狀態(tài)：OnlinePartition。

PartitionLeaderElectionStrategy 分區(qū)Leader選舉策略，這里傳入的是OfflinePartitionLeaderElectionStrategy，分區(qū)離線狀態(tài)的Leader選舉策略

這里判斷一下分區(qū)是否有效的依據(jù)主要是要根據(jù)狀態(tài)機(jī)設(shè)置的驅(qū)動條件，例如只有分區(qū)狀態(tài)為OnlinePartition、NewPartition、OfflinePartition三個狀態(tài)才能轉(zhuǎn)換為OnlinePartition。

接下來重點(diǎn)看變更為OnlinePartition的具體實(shí)現(xiàn)邏輯，具體代碼如下所示：

具體實(shí)現(xiàn)分為3個步驟：

首先先分別帥選出當(dāng)前狀態(tài)為NewPartition的集合與（OfflinePartition或者OnlinePartition）分區(qū)。

狀態(tài)為NewPartition的分區(qū)，執(zhí)行分區(qū)的初始化，通常為分區(qū)擴(kuò)容或主題新創(chuàng)建

狀態(tài)為OfflinePartition或者OnlinePartition的執(zhí)行分區(qū)重新選舉，因?yàn)檫@些集合中的分區(qū)是當(dāng)前沒有Leader的分區(qū)，這些分區(qū)暫時無法接受讀寫請求。

接下來我們重點(diǎn)看一下離線狀態(tài)變更為OnlinePartition的分區(qū)leader選舉實(shí)現(xiàn)，具體方法為：PartitionStateMachine的electLeaderForPartitions方法，其代碼如下所示：

這個方法的實(shí)現(xiàn)結(jié)構(gòu)比較簡單，返回值為兩個集合，一個選舉成功的集合，一個選舉失敗的集合，同時選舉過程中如果出現(xiàn)可恢復(fù)異常，則會進(jìn)行重試。

具體的重試邏輯由doElectLeaderForPartitions方法實(shí)現(xiàn)，該方法非常復(fù)雜。

3.2 分區(qū)選舉機(jī)制

分區(qū)選舉由PartitionStateMachine的doElectLeaderForPartitions方法實(shí)現(xiàn)，接下來分步進(jìn)行講解。

Step1：首先從Zookeeper中獲取需要選舉分區(qū)的元信息，代碼如下所示：

Kafka中主題的路由信息存儲在Zookeeper中，具體路徑為：/{namespace}/brokers/topics/{topicName}}/partitions/{partition}/state,具體存儲的內(nèi)容如下所示：

Step2：將查詢出來的主題分區(qū)元信息，組裝成Map< TopicPartition, LeaderIsrAndControllerEpoch>的Map結(jié)構(gòu)，代碼如下所示：

Step3：將分區(qū)中的controllerEpoch與當(dāng)前Kafka Controller的epoch對比，刷選出無效與有效集合，具體代碼如下所示：

如果當(dāng)前控制器的controllerEpoch小于分區(qū)狀態(tài)中的controllerEpoch,說明已有新的Broker已取代當(dāng)前Controller成為集群新的Controller，本次無法進(jìn)行Leader選取，并且打印日志。

Step4：根據(jù)Leader選舉策略進(jìn)行Leader選舉，代碼如下所示:

由于我們這次是由OfflinePartition狀態(tài)向OnlinePartition狀態(tài)轉(zhuǎn)換，進(jìn)入的分支為leaderForOffline，稍后我們再詳細(xì)介紹該方法，經(jīng)過選舉后的返回值為兩個集合，其中partitionsWithoutLeaders表示未成功選舉出Leader的分區(qū)，而partitionsWithLeaders表示成功選舉出Leader的分區(qū)。

Step5:沒有成功選舉出Leader的分區(qū)打印對應(yīng)日志，并加入到失敗隊(duì)列集合中，如下圖所示：

Step5:將選舉結(jié)果更新到zookeeper中，如下圖所示：

Step6：將最新的分區(qū)選舉結(jié)果同步到其他Broker節(jié)點(diǎn)上。

更新分區(qū)狀態(tài)的請求LEADER_AND_ISR被其他Broker接受后，會根據(jù)分區(qū)的leader與副本信息，成為該分區(qū)的Leader節(jié)點(diǎn)或從節(jié)點(diǎn)，關(guān)于這塊的實(shí)現(xiàn)細(xì)節(jié)在專欄的后續(xù)文章中會專門提及。

那OfflinePartitionLeaderElectionStrategy選舉策略具體是如何進(jìn)行選舉的呢？接下來我們探究其實(shí)現(xiàn)細(xì)節(jié)。

3.3 OfflinePartitionLeaderElectionStrategy選舉策略

OfflinePartitionLeaderElectionStrategy的選舉策略實(shí)現(xiàn)代碼見PartitionStateMachine的leaderForOffline，我們還是采取分步探討的方式。

Step1：主要初始化幾個集合，代碼如下

對上面的變量做一個簡單介紹：

partitionsWithNoLiveInSyncReplicas 分區(qū)的副本所在的Broker全部不存活

partitionsWithLiveInSyncReplicas 分區(qū)副本集合所在的broker部分或全部存活

partitionsWithUncleanLeaderElectionState 主題是否開啟了副本不在isr集合中也可以參與Leader競選，可在主題級別設(shè)置unclean.leader.election.enable，默認(rèn)為false。

Step2：執(zhí)行分區(qū)Leader選舉，具體實(shí)現(xiàn)代碼如下所示：

首先解釋如下幾個變量的含義：

assignment 分區(qū)設(shè)置的副本集（所在brokerId）。

liveReplicas 當(dāng)前在線的副本集。

具體的選舉算法如下所示：

離線轉(zhuǎn)在線的選舉算法比較簡單：如果unclean.leader.election.enable=false，則從存活的ISR集合中選擇第一個成為分區(qū)的Leader，如果沒有存活的ISR副本，并且unclean.leader.election.enable=true，則選擇一個在線的副本，否則返回NONE，表示沒有成功選擇一個合適的Leader。

然后返回本次選舉的結(jié)果，完成本次選舉。

4、總結(jié)

本文從一個生產(chǎn)實(shí)際故障開始進(jìn)行分析，經(jīng)過分析得出單副本主題在集群中單臺節(jié)點(diǎn)下線會引起部分隊(duì)列無法寫入，解決辦法是要先執(zhí)行主題分區(qū)移動，也就是將需要停止的broker上所在的分區(qū)移動到其他broker上，這個過程并不會對消息發(fā)送，消息消費(fèi)造成影響。