Kafka 如何基于 KRaft 實現(xiàn)集群最終一致性協(xié)調(diào)

作者：AutoMQ 2024-06-04 09:51:48

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

一、架構概覽

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

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

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

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

圖片

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

運維部署：3 個 ZK 節(jié)點；2..N 個 Broker 節(jié)點，其中一個 Broker 承擔 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 負責全局的協(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é)點可以同時承擔 Controller 和 Broker 兩個角色，因此一套最小生產(chǎn)集群只需要 3 個節(jié)點。在測試環(huán)境更可以只以 1 節(jié)點模式就可以輕量地拉起一個 Kafka 集群。
通信協(xié)調(diào)：Controller 節(jié)點底層通過 Raft 協(xié)議達成一致，Controller 的內(nèi)存狀態(tài)通過 #replay Raft Log 來構建，因此 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ù)，請求處理流程簡要如下：

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

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

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

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

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

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

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

圖片

相比簡版的處理時序：

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

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

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

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

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

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

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

2.2 Broker

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

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

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

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

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

三、總結

參考資料

[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

責任編輯：武曉燕來源： AutoMQ

Kafka ZK 機制