一、架構概覽   

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

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

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

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

圖片

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

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

在 KRaft 模式下：

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

Raft 的原理和實現已經有很多優秀的文章介紹過了，就不在此贅述了。下面著重介紹一下 Kafka 如何基于 KRaft 實現集群的最終一致性協調。

二、最終一致性協調

最終一致性協調分為兩部分：Controller 內存數據與 KRaft 的一致性；Broker （分區 / 配置 / ...）狀態與期望的一致性。

2.1 Controller

Controller 在生產環境中通常由 3 個節點組成 Quorum，底層使用 KRaft 來進行一致性協調，KRaft 的 Leader 即是 Controller Leader。

只有 Leader 會進行請求處理，Follower 只會跟隨 Replay KRaft 中的數據，請求處理流程簡要如下：

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

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

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

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

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

通過上述的處理時序，Controller 就可以做到“內存狀態與 KRaft ”和“多節點之間的內存狀態”的一致性：

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

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

圖片

相比簡版的處理時序：

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

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

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

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

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

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

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

2.2 Broker

在上一章節我們提到，Controller Follower 會 #replay KRaft 中的數據來構建自己的內存狀態。Broker 同理也一樣會訂閱 KRaft 中的 Records 來構建自己的內存元數據，并且根據這些 Records 來執行特定的變更。

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

1. Controller 處理分區移動請求，并生成 PartitionChangeRecord{P1=B2}提交到 KRaft；
2. B1 #replay到對應的變更記錄，更新內存元數據記錄 P1 在 B2 上，并開始關閉 P1；
3. B2#replay到對應的變更記錄，更新內存元數據記錄 P1 在 B2 上，并開始打開 P1；

這時候 B1 和 B2 都可以通過內存元數據提供一致的的 Topic Metadata 查詢服務，并且完成了分區 P1 的移動。

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

三、總結   

我們可以看出 KRaft 替換 ZK，并不是元數據存儲重新造輪子，而核心是集群協調機制的演進。整個通信協調機制本質上是事件驅動模型，也就是 Metadata as an Event Log，Leader 通過 KRaft 生產權威的事件，Follower 和 Broker 通過監聽 KRaft 來獲得這些事件，并且順序處理事件，達到集群狀態和期望的最終一致。

參考資料

[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