一、架構(gòu)概覽   

Zookeeper 提供了配置服務(wù)、分布式同步、命名服務(wù)、Leader 選舉和集群管理等功能，在大數(shù)據(jù)時代的開始很多開源產(chǎn)品都依賴 Zookeeper 來構(gòu)建，Apache Kafka 也不例外。但是隨著 Kafka 功能的演進和應(yīng)用的場景越來越多：

• 基于 Zookeeper 的協(xié)作模式，使得 Kafka 的集群一致性維護越來越復(fù)雜；
• 受到 Zookeeper 性能的限制，使得 Kafka 無法支撐更大的集群規(guī)模；
• 并且 Zookeeper 自身帶來的運維復(fù)雜性和產(chǎn)品穩(wěn)定性，也同樣將復(fù)雜度和風(fēng)險負(fù)擔(dān)傳遞到 Kafka 運維人員；

因此作為 Zookeeper 的替代，Kafka 3.3.1 提供了 KRaft 元數(shù)據(jù)管理組件。

下圖來自于 KIP-500 [1]提案，左右分別是 Zookeeper 模式和 KRaft 模式的部署架構(gòu)圖。

圖片

在 Zookeeper （后面簡稱為 ZK）模式下：

• 運維部署：3 個 ZK 節(jié)點；2..N 個 Broker 節(jié)點，其中一個 Broker 承擔(dān) Controller 的角色。除了拉起一套最小生產(chǎn)的 Kafka 集群需要至少 3 + N 的資源外，Kafka 的運維人員要同時掌握 ZK 和 Kafka Broker 兩套完全不同的系統(tǒng)的運維方式。
• 通信協(xié)調(diào)：ZK 節(jié)點之間通過 ZAB 協(xié)議進行一致性協(xié)調(diào)；Broker 會通過 ZK 來選出一個 Controller 負(fù)責(zé)全局的協(xié)調(diào)，同時也會直接修改 ZK 里的數(shù)據(jù)；Controller 也會監(jiān)聽和修改 ZK 里的數(shù)據(jù)，并調(diào)用 Broker 來完成集群的協(xié)調(diào)。雖然 ZK 之間的一致性由 ZAB 來保障了，但是 ZK 與 Controller 之間和 Controller 與 Broker 之間的一致性是相對比較脆弱的。

在 KRaft 模式下：

• 運維部署：3 個 Controller 節(jié)點；0..N 個 Broker 節(jié)點。Kafka 節(jié)點可以同時承擔(dān) Controller 和 Broker 兩個角色，因此一套最小生產(chǎn)集群只需要 3 個節(jié)點。在測試環(huán)境更可以只以 1 節(jié)點模式就可以輕量地拉起一個 Kafka 集群。
• 通信協(xié)調(diào)：Controller 節(jié)點底層通過 Raft 協(xié)議達(dá)成一致，Controller 的內(nèi)存狀態(tài)通過 #replay Raft Log 來構(gòu)建，因此 Controller 之間的內(nèi)存狀態(tài)都是一致的；Broker 訂閱 KRaft Log 維護和 Controller 一致的內(nèi)存狀態(tài)，并且通過事件驅(qū)動的方式執(zhí)行 Partition Reassignment 之類的操作來實現(xiàn)集群最終一致性協(xié)調(diào)。整個集群的狀態(tài)維護和一致性協(xié)調(diào)都是基于 KRaft 中的事件。

Raft 的原理和實現(xiàn)已經(jīng)有很多優(yōu)秀的文章介紹過了，就不在此贅述了。下面著重介紹一下 Kafka 如何基于 KRaft 實現(xiàn)集群的最終一致性協(xié)調(diào)。

二、最終一致性協(xié)調(diào)

最終一致性協(xié)調(diào)分為兩部分：Controller 內(nèi)存數(shù)據(jù)與 KRaft 的一致性；Broker （分區(qū) / 配置 / ...）狀態(tài)與期望的一致性。

2.1 Controller

Controller 在生產(chǎn)環(huán)境中通常由 3 個節(jié)點組成 Quorum，底層使用 KRaft 來進行一致性協(xié)調(diào)，KRaft 的 Leader 即是 Controller Leader。

只有 Leader 會進行請求處理，F(xiàn)ollower 只會跟隨 Replay KRaft 中的數(shù)據(jù)，請求處理流程簡要如下：

1. 當(dāng) Leader 網(wǎng)絡(luò)層接收到 Broker 發(fā)來的請求后，會將請求首先放入到事件隊列中，由后臺的單線程來處理事件隊列中的請求。通過單線程處理機制簡化了并發(fā)編程的復(fù)雜度，并且確保所有請求可以順序處理；
2. 單線程處理器運行請求對應(yīng)的 Manager 邏輯。Manager 根據(jù)當(dāng)前內(nèi)存中維護的狀態(tài)，生成響應(yīng)和變更的 Records；
3. 最后再把變更的 Records 提交到 KRaft 中，等多數(shù)派確認(rèn)后就可以將響應(yīng)返回，并 #replay(Records) 修改 Manager 維護的內(nèi)存狀態(tài)；
4. 同時 Follower 也會將 KRaft 中的 Records #replay到內(nèi)存中，內(nèi)存數(shù)據(jù)持續(xù)的保持同步；

以 CAS(expectValue, newValue) 舉例說明上述的流程，假設(shè)內(nèi)存中的初始狀態(tài)為 1，Broker Client 提交了請求 CAS(1, 2) 到 Controller：

1. 首先 Leader 會將請求放到事件隊列中；
2. 然后 Manager 以單線程模式處理請求，判斷內(nèi)存中的值是 1，等于請求的 expectValue，因此生成成功響應(yīng)和 Record{value = 2};
3. 最后再把變更的 Records 提交到 KRaft 中，KRaft 確認(rèn)后返回給請求方響應(yīng)，并將 Record{value = 2} replay 到 Manager，Manager 內(nèi)存狀態(tài)更新為 2；

簡而言之，Controller 簡版的處理時序如下：

開始處理請求 A -> Manager 生成響應(yīng)和 Records -> Records 在 KRaft 多數(shù)派確認(rèn) -> Manager#replay(Records) -> 返回響應(yīng) -> 處理下一條請求...

通過上述的處理時序，Controller 就可以做到“內(nèi)存狀態(tài)與 KRaft ”和“多節(jié)點之間的內(nèi)存狀態(tài)”的一致性：

• 內(nèi)存狀態(tài)與 KRaft ：Controller 的內(nèi)存狀態(tài)都是基于 KRaft 確認(rèn)的 Records 變更 #replay出來的，因此內(nèi)存狀態(tài)和 KRaft 保持一致；
• 多節(jié)點之間的內(nèi)存狀態(tài)：KRaft 底層保證了多節(jié)點的 KRaft Log 是一致的，然后基于 “內(nèi)存狀態(tài)與 KRaft” 的一致性，通過傳遞性原則，因此多節(jié)點之間的內(nèi)存狀態(tài)也是一致的；

Controller 簡版的處理時序在正確性上沒什么問題，但在性能上有所瓶頸。假設(shè)每次 KRaft 多數(shù)派確認(rèn)需要 2ms，意味著 Controller 處理請求的最大吞吐為 500 req/s。因此 Kafka 的實際處理模型中將最耗時的 KRaft 確認(rèn)這步從處理時序中移除了。具體流程如下圖所示：

圖片

相比簡版的處理時序：

• Leader 的 Manager 產(chǎn)生出 Records 后立刻 #replay 更新內(nèi)存狀態(tài)，并異步提交 Records 到 KRaft，這時候就可以繼續(xù)處理下一個請求了；
• 響應(yīng)仍舊是 KRaft 多數(shù)派確認(rèn)后再返回；
• Follower 的內(nèi)存狀態(tài)仍舊是從 KRaft Log 的 Records #replay 更新；

Controller 處理請求的最大吞吐為：Min(1s / Manager 代碼執(zhí)行 CPU 耗時, KRaft 寫入吞吐)。

然而先 #replay 到內(nèi)存再讓 KRaft 確認(rèn)可能會造成內(nèi)存里面有臟數(shù)據(jù)，仍舊以 CAS(1, 2) 舉例，考慮如下場景：

1. Controller Leader 的 Manager 通過 #replay 將內(nèi)存值從 1 更新成 2；
2. Leader 提交 Record{value=2}到 KRaft；
3. 假設(shè)這時候由于心跳超時抖動等原因，導(dǎo)致該節(jié)點不再是 KRaft Leader 了，這時候會提交失敗，返回客戶端失?。?/li>
4. 這時 Controllers 節(jié)點內(nèi)存中的狀態(tài)分別為 2、1、1，KRaft 中的狀態(tài)為 1，集群狀態(tài)不一致；

為了解決這個問題，Kafka 設(shè)計了一系列支持 MVCC 的 Timeline 數(shù)據(jù)結(jié)構(gòu)：TimelineHashMap、TimelineHashSet、TimelineInteger、TimelineLong 和底層的 SnapshotRegistry。Controller 的內(nèi)存狀態(tài)都通過 Timeline 數(shù)據(jù)結(jié)構(gòu)來維護，當(dāng)出現(xiàn) Leader 切換時，舊的 Leader 會將 Timeline 數(shù)據(jù)結(jié)構(gòu)的數(shù)據(jù)回滾到上一個已經(jīng)被 KRaft 多數(shù)派確認(rèn)的狀態(tài)，來保證舊 Leader 內(nèi)存中不會有臟數(shù)據(jù)。

可能細(xì)心的小伙伴會發(fā)現(xiàn)，解決了寫入的臟數(shù)據(jù)問題，那是不是可能讀到還未被 KRaft 確認(rèn)的數(shù)據(jù)呢？Timeline 數(shù)據(jù)結(jié)構(gòu)也考慮到了這點，例如 TimelineLong 提供了 #get(epoch) 接口，其中 epoch 通常傳入的是 KRaft CommitedOffset，以此來保障讀到的數(shù)據(jù)都是 KRaft 確認(rèn)過的數(shù)據(jù)。

對 Timeline 數(shù)據(jù)結(jié)構(gòu)有興趣的小伙伴，可以自行研究一下 server-common 模塊下 org.apache.kafka.timeline 這個包的實現(xiàn)。

2.2 Broker

在上一章節(jié)我們提到，Controller Follower 會 #replay KRaft 中的數(shù)據(jù)來構(gòu)建自己的內(nèi)存狀態(tài)。Broker 同理也一樣會訂閱 KRaft 中的 Records 來構(gòu)建自己的內(nèi)存元數(shù)據(jù)，并且根據(jù)這些 Records 來執(zhí)行特定的變更。

以分區(qū)管理為例，假設(shè)集群有 B1 和 B2 兩個節(jié)點，用戶將分區(qū) P1 從 B1 移動到 B2（簡化 ISR 變更的過程）：

1. Controller 處理分區(qū)移動請求，并生成 PartitionChangeRecord{P1=B2}提交到 KRaft；
2. B1 #replay到對應(yīng)的變更記錄，更新內(nèi)存元數(shù)據(jù)記錄 P1 在 B2 上，并開始關(guān)閉 P1；
3. B2#replay到對應(yīng)的變更記錄，更新內(nèi)存元數(shù)據(jù)記錄 P1 在 B2 上，并開始打開 P1；

這時候 B1 和 B2 都可以通過內(nèi)存元數(shù)據(jù)提供一致的的 Topic Metadata 查詢服務(wù)，并且完成了分區(qū) P1 的移動。

通過這種方式，很多變更 Controller 無需再主動調(diào)用 Broker 的 RPC 來嘗試將集群推進到某個狀態(tài)，也無需處理 RPC 調(diào)用中的順序和冪等重試等問題。轉(zhuǎn)換思路，Controller 通過 KRaft 來下發(fā)期望的狀態(tài)，然后 Broker 去達(dá)成狀態(tài)，這和 K8s 推薦的聲明式管理有異曲同工之妙。

三、總結(jié)   

我們可以看出 KRaft 替換 ZK，并不是元數(shù)據(jù)存儲重新造輪子，而核心是集群協(xié)調(diào)機制的演進。整個通信協(xié)調(diào)機制本質(zhì)上是事件驅(qū)動模型，也就是 Metadata as an Event Log，Leader 通過 KRaft 生產(chǎn)權(quán)威的事件，F(xiàn)ollower 和 Broker 通過監(jiān)聽 KRaft 來獲得這些事件，并且順序處理事件，達(dá)到集群狀態(tài)和期望的最終一致。

參考資料

[1] KIP-500 Replace Zookeeper with a Self-Managed Metadata Quorum：https://cwiki.apache.org/confluence/display/KAFKA/KIP-500%3A+Replace+ZooKeeper+with+a+Self-Managed+Metadata+Quorum

[2] Timeline：https://github.com/apache/kafka/tree/trunk/server-common/src/main/java/org/apache/kafka/timeline