Kafka Controller 是 Kafka 的核心組件，在前面的文章中，已經(jīng)詳細(xì)講述過(guò) Controller 部分的內(nèi)容。在過(guò)去的幾年根據(jù)大家在生產(chǎn)環(huán)境中應(yīng)用的反饋，Controller 也積累了一些比較大的問(wèn)題，而針對(duì)這些問(wèn)題的修復(fù)，代碼的改動(dòng)量都是非常大的，無(wú)疑是一次重構(gòu)，因此，社區(qū)準(zhǔn)備在新版的系統(tǒng)里對(duì) Controller 做一些相應(yīng)的優(yōu)化（0.11.0及以后的版本），相應(yīng)的設(shè)計(jì)方案見(jiàn)：Kafka Controller Redesign，本文的內(nèi)容就是結(jié)合這篇文章做一個(gè)簡(jiǎn)單的總結(jié)。

Controller 功能

在一個(gè) Kafka 中，Controller 要處理的事情總結(jié)如下表所示：

Controller 目前存在的問(wèn)題

之所以要重新設(shè)計(jì) Controller，是因?yàn)楝F(xiàn)在的 Controller 積累了一些比較難解決的問(wèn)題，這些問(wèn)題解決起來(lái)，代碼改動(dòng)量都是巨大的，甚至需要改變 controller 部門的設(shè)計(jì)，基本就跟重構(gòu)差不多了，下面我們先來(lái)了看一下 controller 之前（主要是 0.11.0 之前的版本）存在的一些問(wèn)題。

目前遇到的比較大的問(wèn)題有以下幾個(gè)：

1. Partition 級(jí)別同步 zk 寫；
2. sequential per-partition controller-to-broker requests；
3. Controller 復(fù)雜的并發(fā)語(yǔ)義；
4. 代碼組織混亂；
5. 控制類請(qǐng)求與數(shù)據(jù)類請(qǐng)求未分離；
6. Controller 給 broker 的請(qǐng)求中沒(méi)有 broker 的 generation信息；
7. ZkClient 阻礙 Client 的狀態(tài)管理。

Partition 級(jí)別同步 zk 寫

zookeeper 的同步寫意味著在下次寫之前需要等待前面整個(gè)過(guò)程的結(jié)束，而且由于它們都是 partition 粒度的（一個(gè) Partition 一個(gè) Partition 的去執(zhí)行寫操作），對(duì)于 Partition 非常多的集群來(lái)說(shuō)，需要等待的時(shí)間會(huì)更長(zhǎng)，Controller 通常會(huì)在下面這兩個(gè)地方做 Partition 級(jí)別 zookeeper 同步寫操作：

1. PartitionStateMachine 在進(jìn)行觸發(fā) leader 選舉（partition 目的狀態(tài)是 OnlinePartition），將會(huì)觸發(fā)上面的操作；
2. ReplicaStateMachine 更新 LeaderAndIsr 信息到 zk（replica 狀態(tài)轉(zhuǎn)變?yōu)?OfflineReplica），這種情況也觸發(fā)這種情況，它既阻礙了 Controller 進(jìn)程，也有可能會(huì) zk 造成壓力。

sequential per-partition controller-to-broker requests

Controller 在向 Broker 發(fā)送請(qǐng)求，有些情況下也是 Partition 粒度去發(fā)送的，效率非常低，比如在 Controller 處理 broker shutdown 請(qǐng)求時(shí)，這里是按 Partition 級(jí)別處理，每處理一個(gè) Partition 都會(huì)執(zhí)行 Partition、Replica 狀態(tài)變化以及 Metadata 更新，并且調(diào)用?sendRequestsToBrokers()?向 broker 發(fā)送請(qǐng)求，這樣的話，效率將變得非常低。

Controller 復(fù)雜的并發(fā)語(yǔ)義

Controller 需要在多個(gè)線程之間共享狀態(tài)信息，這些線程有：

1. IO threads handling controlled shutdown requests
2. The ZkClient org.I0Itec.zkclient.ZkEventThread processing zookeeper callbacks sequentially；
3. The TopicDeletionManager kafka.controller.DeleteTopicsThread；
4. Per-broker RequestSendThread within ControllerChannelManager.

所有這些線程都需要訪問(wèn)或修改狀態(tài)信息（ControllerContext），現(xiàn)在它們是通過(guò) ControllerContext 的 controllerLock（排它鎖）實(shí)現(xiàn)的，Controller 的并發(fā)變得虛弱無(wú)力。

代碼組織混亂

KafkaController 部分的代碼組織（KafkaController、PartitionStateMachine 和 ReplicaStateMachine）不是很清晰，比如，下面的問(wèn)題就很難回答：

1. where and when does zookeeper get updated?
2. where and when does a controller-to-broker request get formed?
3. what impact does a failing zookeeper update or controller-to-broker request have on the cluster state?

這也導(dǎo)致了這部分很多開(kāi)發(fā)者不敢輕易去改動(dòng)。

控制類請(qǐng)求與數(shù)據(jù)類請(qǐng)求未分離

現(xiàn)在 broker 收到的請(qǐng)求，有來(lái)自 client、broker 和 controller 的請(qǐng)求，這些請(qǐng)求都會(huì)被放到同一個(gè) requestQueue 中，它們有著同樣的優(yōu)先級(jí)，所以來(lái)自 client 的請(qǐng)求很可能會(huì)影響來(lái)自 controller 請(qǐng)求的處理（如果是 leader 變動(dòng)的請(qǐng)求，ack 設(shè)置的不是 all，這種情況有可能會(huì)導(dǎo)致數(shù)據(jù)丟失）。

Controller 給 broker 的請(qǐng)求中沒(méi)有 broker 的 generation信息

這里的 Broker generation 代表著一個(gè)標(biāo)識(shí)，每當(dāng)它重新加入集群時(shí)，這個(gè)標(biāo)識(shí)都會(huì)變化。如果 Controller 的請(qǐng)求沒(méi)有這個(gè)信息的話，可能會(huì)導(dǎo)致一個(gè)重啟的 Broker 收到之前的請(qǐng)求，讓 Broker 進(jìn)入到一個(gè)錯(cuò)誤的狀態(tài)。

比如，Broker 收到之前的 StopReplica 請(qǐng)求，可能會(huì)導(dǎo)致副本同步線程退出。

ZkClient 阻礙 Client 的狀態(tài)管理

這里的狀態(tài)管理指的是當(dāng) Client 發(fā)生重連或會(huì)話過(guò)期時(shí)，Client 可以監(jiān)控這種狀態(tài)變化，并做出一些處理，因?yàn)殚_(kāi)源版的 ZKClient 在處理 notification 時(shí)，是線性處理的，一些 notification 會(huì)被先放到 ZkEventThread’s queue 中，這樣會(huì)導(dǎo)致一些最新的 notification 不能及時(shí)被處理，特別是與 zk 連接斷開(kāi)重連的情況。

Controller 改進(jìn)方案

關(guān)于上述問(wèn)題，Kafka 提出了一些改進(jìn)方案，有些已經(jīng)在最新版的系統(tǒng)中實(shí)現(xiàn)，有的還在規(guī)劃中。

使用異步的 zk API

Zookeeper 的 client 提供三種執(zhí)行請(qǐng)求的方式：

1. 同步調(diào)用，意味著下次請(qǐng)求需要等待當(dāng)前當(dāng)前請(qǐng)求的完成；
2. 異步調(diào)用，意味著不需要等待當(dāng)前請(qǐng)求的完成就可以開(kāi)始下次請(qǐng)求的執(zhí)行，并且我們可以通過(guò)回調(diào)機(jī)制去處理請(qǐng)求返回的結(jié)果；
3. 單請(qǐng)求的 batch 調(diào)用，意味著 batch 內(nèi)的所有請(qǐng)求都會(huì)在一次事務(wù)處理中完成，這里需要關(guān)注的是 zookeeper 的 server 對(duì)單請(qǐng)求的大小是有限制的（jute.maxbuffer）。

文章中給出了三種請(qǐng)求的測(cè)試結(jié)果，Kafka 最后選取的是異步處理機(jī)制，因?yàn)閷?duì)于單請(qǐng)求處理，異步處理更加簡(jiǎn)潔，并且相比于同步處理還可以保持一個(gè)更好的寫性能。

improve controller-to-broker request batching

這個(gè)在設(shè)計(jì)文檔還是 TODO 狀態(tài)，具體的方案還沒(méi)確定，不過(guò)基本可以猜測(cè)一下，因?yàn)槟康氖翘岣?batch 發(fā)送能力，那么只能是在調(diào)用對(duì)每個(gè) broker 的 RequestSenderThread 線程發(fā)送請(qǐng)求之前，做一下檢測(cè)，而不是來(lái)一個(gè)請(qǐng)求立馬就發(fā)送，這是一個(gè)性能與時(shí)間的權(quán)衡，如果不是立馬發(fā)送請(qǐng)求，那么可能會(huì)帶來(lái) broker 短時(shí) metadata 信息的不一致，這個(gè)不一致時(shí)間不同的應(yīng)用場(chǎng)景要求是不一樣的。

單線程的事件處理模型

采用單線程的時(shí)間處理模型將極大簡(jiǎn)化 Controller 的并發(fā)實(shí)現(xiàn)，只允許這個(gè)線程訪問(wèn)和修改 Controller 的本地狀態(tài)信息，因此在 Controller 部分也就不需要到處加鎖來(lái)保證線程安全了。

目前 1.1.0 的實(shí)現(xiàn)中，Controller 使用了一個(gè) ControllerEventThread 線程來(lái)處理所有的 event，目前可以支持13種不同類型事件：

1. Idle：代表當(dāng)前 ControllerEventThread 處理空閑狀態(tài)；
2. ControllerChange：Controller 切換處理；
3. BrokerChange：Broker 變動(dòng)處理，broker 可能有上線或掉線；
4. TopicChange：Topic 新增處理；
5. TopicDeletion：Topic 刪除處理；
6. PartitionReassignment：Partition 副本遷移處理；
7. AutoLeaderBalance：自動(dòng) rebalance 處理；
8. ManualLeaderBalance：最優(yōu) leader 選舉處理，這里叫做手動(dòng) rebalance，手動(dòng)去切流量；
9. ControlledShutdown：優(yōu)雅關(guān)閉 broker；
10. IsrChange：Isr 變動(dòng)處理；
11. LeaderAndIsrResponseReceived；
12. LogDirChange：Broker 某個(gè)目錄失敗后的處理（比如磁盤壞掉等）；
13. ControllerShutdown：ControllerEventThread 處理這個(gè)事件時(shí)，會(huì)關(guān)閉當(dāng)前線程。

重構(gòu)集群狀態(tài)管理

這部分的改動(dòng)，目前社區(qū)也沒(méi)有一個(gè)很好的解決思路，重構(gòu)這部分的目的是希望 Partition、Replica 的狀態(tài)管理變得更清晰一些，讓我們從代碼中可以清楚地明白狀態(tài)是在什么時(shí)間、什么地方、什么條件下被觸發(fā)的。這個(gè)優(yōu)化其實(shí)是跟上面那個(gè)有很大關(guān)聯(lián)，采用單線程的事件處理模型，可以讓狀態(tài)管理也變得更清晰。

prioritize controller requests

我們想要把控制類請(qǐng)求與數(shù)據(jù)類請(qǐng)求分開(kāi)，提高 controller 請(qǐng)求的優(yōu)先級(jí)，這樣的話即使 Broker 中請(qǐng)求有堆積，Broker 也會(huì)優(yōu)先處理控制類的請(qǐng)求。

這部分的優(yōu)化可以在網(wǎng)絡(luò)層的 RequestChannel 中做，RequestChannel 可以根據(jù)請(qǐng)求的 id 信息把請(qǐng)求分為正常的和優(yōu)先的，如果請(qǐng)求是 UpdateMetadataRequest、LeaderAndIsrRequest 或者 StopReplicaRequest，那么這個(gè)請(qǐng)求的優(yōu)先級(jí)應(yīng)該提高。實(shí)現(xiàn)方案有以下兩種：

1. 在請(qǐng)求隊(duì)列中增加一個(gè)優(yōu)先級(jí)隊(duì)列，優(yōu)先級(jí)高的請(qǐng)求放到 the prioritized request queue 中，優(yōu)先級(jí)低的放到普通請(qǐng)求隊(duì)列中，但是無(wú)論使用一個(gè)定時(shí)拉取（poll）還是2個(gè)定時(shí)拉取，都會(huì)帶來(lái)其他的問(wèn)題，要么是增大普通請(qǐng)求的處理延遲，要么是增大了優(yōu)先級(jí)高請(qǐng)求的延遲；
2. 直接使用優(yōu)先級(jí)隊(duì)列代替現(xiàn)在的普通隊(duì)列，設(shè)計(jì)上更傾向與這一種。

目前這部分在1.1.0中還未實(shí)現(xiàn)。

Controller 發(fā)送請(qǐng)求中添加 broker 的 generation 信息

generation 信息是用來(lái)標(biāo)識(shí)當(dāng)前 broker 加入集群 epoch 信息，每當(dāng) broker 重新加入集群中，該 broker.id 對(duì)應(yīng)的 generation 都應(yīng)該變化（要求遞增），目前有兩種實(shí)現(xiàn)方案：

1. 為 broker 分配的一個(gè)全局唯一的 id，由 controller 廣播給其他 broker；
2. 直接使用 zookeeper 的 zxid 信息（broker.id 注冊(cè)時(shí)的 zxid）。

直接使用原生的 Zookeeper client

Client 端的狀態(tài)管理意味著當(dāng) Client 端發(fā)生狀態(tài)變化（像連接中斷或回話超時(shí)）時(shí)，我們有能力做一些操作。其中，zookeeper client 有效的狀態(tài)（目前的 client 比下面又多了幾種狀態(tài)，這里先不深入）是:

• NOT_CONNECTED： the initial state of the client；
• CONNECTING： the client is establishing a connection to zookeeper；
• CONNECTED： the client has established a connection and session to zookeeper；
• CLOSED： the session has closed or expired。

有效的狀態(tài)轉(zhuǎn)移是：

• NOT_CONNECTED > CONNECTING
• CONNECTING > CONNECTED
• CONNECTING > CLOSED
• CONNECTED > CONNECTING
• CONNECTED > CLOSED

最開(kāi)始的設(shè)想是直接使用原生 Client 的異步調(diào)用方式，這樣的話依然可以通過(guò)回調(diào)方法監(jiān)控到狀態(tài)的變化（像連接中斷或回話超時(shí)），同樣，在每次事件處理時(shí)，可以通過(guò)檢查狀態(tài)信息來(lái)監(jiān)控到 Client 狀態(tài)的變化，及時(shí)做一些處理。

當(dāng)一個(gè) Client 接收到連接中斷的 notification（Client 狀態(tài)變成了 CONNECTING 狀態(tài)），它意味著 Client 不能再?gòu)?zookeeper 接收到任何 notification 了。如果斷開(kāi)連接，對(duì)于 Controller 而言，無(wú)論它現(xiàn)在正在做什么它都應(yīng)該先暫停，因?yàn)榭赡芗�群�?Controller 已經(jīng)切換到其他機(jī)器上了，只是它還沒(méi)接收到通知，它如果還在工作，可能會(huì)導(dǎo)致集群狀態(tài)不一致。當(dāng)連接斷開(kāi)后，Client 可以重新建立連接（re-establish，狀態(tài)變?yōu)?CONNECTED）或者會(huì)話過(guò)期（狀態(tài)變?yōu)?CLOSED，會(huì)話過(guò)期是由 zookeeper Server 來(lái)決定的）。如果變成了 CONNECTED 狀態(tài)，Controller 應(yīng)該重新開(kāi)始這些暫停的操作，而如果狀態(tài)變成了 CLOSED 狀態(tài)，舊的 Controller 就會(huì)知道它不再是 controller，應(yīng)該丟棄掉這些任務(wù)。

參考

• Kafka Controller Redesign；
• Kafka controller重設(shè)計(jì)。

 

標(biāo)簽： 安全 代碼 開(kāi)發(fā)者 網(wǎng)絡(luò)

版權(quán)申明：本站文章部分自網(wǎng)絡(luò)，如有侵權(quán)，請(qǐng)聯(lián)系：west999com@outlook.com
特別注意：本站所有轉(zhuǎn)載文章言論不代表本站觀點(diǎn)！
本站所提供的圖片等素材，版權(quán)歸原作者所有，如需使用，請(qǐng)與原作者聯(lián)系。