Controller元數據：Controller都保存有哪些東西？有幾種狀態？
今天我們進入到Kafka源碼解析的第三大模塊——控制器（Controller）的學習。作為Kafka集群中最為關鍵的組件，Controller負責管理集群元數據、協調副本選舉，并且在系統故障時執行恢復策略。本文我們將從Controller的元數據入手，探討它保存的內容、相關源碼解析以及其中的幾種關鍵狀態。
一、Controller的核心職責
在Kafka集群中，Controller承擔的職責至關重要，主要包括：
- 選舉分區的Leader副本：每個Kafka分區都有多個副本，其中一個副本是主副本（Leader），其余副本是跟隨者（Follower）。Controller負責在每個分區發生副本故障時選舉新的Leader。
- 管理集群的元數據：Controller保存著Kafka集群的所有主題、分區、Broker以及副本的相關元數據。
- 同步元數據到其他Broker：當元數據發生變化時，Controller會通知集群中的其他Broker進行同步。
要理解Controller是如何履行這些職責的，首先需要深入理解它管理的元數據和狀態。
二、Controller元數據概覽
在Kafka的Controller中，有一系列元數據用來記錄集群的當前狀態。這些元數據包括：
- ControllerContext：保存Controller的上下文信息。
- ControllerStats：用于統計Controller的性能指標。
- offlinePartitionCount：記錄當前離線的分區數量。
- shuttingDownBrokerIds：保存正在關閉的Broker ID列表。
- liveBrokers：記錄當前存活的Broker信息。
- liveBrokerEpochs：保存每個Broker的epoch值。
- epoch & epochZkVersion：Controller的epoch和Zookeeper版本號。
- allTopics：集群中所有主題的列表。
- partitionAssignments：每個主題分區的副本分配情況。
2.1 ControllerContext
ControllerContext是Controller模塊中至關重要的一個類，負責保存集群的元數據信息。它包含以下核心字段：
public class ControllerContext {
// 當前存活的broker列表
val liveBrokers = mutable.Set[Broker]()
// 各broker的epoch值，記錄了每個broker在Zookeeper的最新狀態
val liveBrokerEpochs = mutable.Map[Broker, Long]()
// 集群中的所有分區
val partitions = mutable.Set[TopicPartition]()
// 每個分區的領導副本信息
val partitionLeadershipInfo = mutable.Map[TopicPartition, LeaderAndIsr]()
// 正在關閉的broker
val shuttingDownBrokerIds = mutable.Set[Int]()
}2.2 liveBrokers & liveBrokerEpochs
這兩個字段保存了當前存活的Broker信息及其對應的epoch。epoch是Kafka中的一個重要概念，它用來標識Broker的變更次數。當一個Broker重啟或狀態發生變化時，它的epoch值會增加，以便Controller能夠判斷Broker狀態是否過期。
// 更新存活的broker信息
def updateLiveBrokers(brokers: Seq[Broker]) = {
liveBrokers.clear()
liveBrokers ++= brokers
}
// 獲取broker的epoch
def brokerEpoch(broker: Broker): Long = {
liveBrokerEpochs.getOrElse(broker, -1)
}在此代碼片段中，我們看到了Controller如何更新和獲取Broker的狀態。liveBrokers集合存儲了當前所有在線的Broker，liveBrokerEpochs則為每個Broker保存了它的最新epoch信息。
2.3 epoch & epochZkVersion
epoch和epochZkVersion是Controller的重要元數據，它們決定了Controller是否處于最新狀態。當Controller的選舉發生時，epoch會遞增，Zookeeper通過epochZkVersion來確保元數據的一致性。
var epoch = -1
var epochZkVersion = -1
// 從Zookeeper獲取最新的Controller epoch
def getEpochFromZookeeper(): Int = {
val zkEpochPath = "/controller_epoch"
val zkData = zkClient.readData(zkEpochPath, new Stat())
val (epochValue, version) = (new String(zkData.data).toInt, zkData.getVersion())
epoch = epochValue
epochZkVersion = version
epoch
}每當Controller獲取Zookeeper上的epoch數據時，它會更新自身的epoch和epochZkVersion，以確保操作的原子性和安全性。
三、Controller的狀態管理
Kafka Controller的狀態管理也非常關鍵，它通過ControllerStats來監控自己的健康狀況和性能表現，確保可以快速檢測并應對潛在的集群問題。
3.1 ControllerStats
ControllerStats主要用于統計Controller的一些性能指標，比如選舉Leader的次數和處理延遲。這些數據對運維Kafka集群非常重要，可以幫助我們快速發現和解決問題。
class ControllerStats {
private val leaderElectionRate = new Meter()
private val offlinePartitionsCount = new AtomicInteger()
// 記錄leader選舉的速率
def markLeaderElection() = {
leaderElectionRate.mark()
}
// 獲取當前離線的分區數量
def offlinePartitionCount: Int = offlinePartitionsCount.get()
// 設置離線分區數量
def setOfflinePartitionCount(count: Int) = {
offlinePartitionsCount.set(count)
}
}在上述代碼片段中，ControllerStats保存了leaderElectionRate（Leader選舉速率）以及offlinePartitionsCount（離線分區數量），這些數據可以通過Kafka的JMX接口導出，供外部監控系統使用。
3.2 offlinePartitionCount
offlinePartitionCount字段記錄了集群中當前處于離線狀態的分區數量。對于Kafka來說，分區的離線意味著無法提供服務，可能會導致消息的不可用或丟失，因此它是Controller非常關心的一個指標。
// 更新離線分區的數量
def updateOfflinePartitionCount(newCount: Int): Unit = {
controllerStats.setOfflinePartitionCount(newCount)
}當集群中有分區進入離線狀態時，Controller會調用updateOfflinePartitionCount方法來更新這個值。
四、Leader選舉與副本管理
接下來，我們來看看Kafka Controller中最為重要的功能之一——Leader選舉。Leader選舉發生在Kafka分區的Leader副本失效時，Controller需要為該分區選擇一個新的Leader。
4.1 Leader選舉過程
Leader選舉的核心邏輯可以簡化為以下步驟：
- 判斷當前分區的Leader是否可用：如果不可用，則進入選舉流程。
- 從所有副本中選擇新的Leader：優先選擇處于"同步副本集合"（ISR）中的副本作為Leader。
- 更新元數據并通知其他Broker：更新Leader信息，向其他Broker廣播新的Leader元數據。
def electLeader(partition: TopicPartition) = {
val isr = controllerContext.partitionLeadershipInfo(partition).isr
val newLeader = selectLeaderFromISR(isr)
updateLeader(partition, newLeader)
broker.notifyLeaderChange(partition, newLeader)
}在這個簡化的代碼片段中，electLeader方法首先從ISR集合中選擇新的Leader，然后調用updateLeader更新Leader信息，并通知集群中的其他Broker。
4.2 副本管理與分區分配
在Kafka集群中，每個主題的分區會被分配給多個Broker，而每個分區又會有多個副本（包括一個Leader和若干個Follower）。Controller通過partitionAssignments字段來管理所有主題分區的副本分配情況。
// 獲取指定主題的分區副本分配信息
def getPartitionAssignments(topic: String): Seq[Int] = {
controllerContext.partitionAssignments(topic)
}partitionAssignments保存了每個主題的分區及其對應的副本信息，這個數據結構對于分區的Leader選舉和數據同步至關重要。
五、Controller的幾種關鍵狀態
在Kafka Controller中，常見的幾種狀態包括：
- 正常運行狀態：Controller正常工作，監控集群狀態，執行Leader選舉和元數據同步。
- 失效狀態：當Controller失去Zookeeper連接或網絡分區時，可能會進入失效狀態。
- 選舉狀態：當Zookeeper中的Controller變更時，新的Controller會進入選舉狀態，競爭成為主Controller。
5.1 Controller失效與重新選舉
當Controller失效時，Zookeeper會觸發一個新的選舉過程，新的Broker可能會成為Controller。以下是Controller進入失效狀態的部分代碼：
def resign(): Unit = {
// 通知其他broker，當前controller不再管理集群
leader
Elector.resign()
// 清空controller上下文
controllerContext.clear()
}resign方法會在Controller失效時被調用，它會清空Controller的上下文信息，并通知其他Broker進行重新選舉。
六、總結
通過本文的講解，我們深入探討了Kafka中Controller的元數據和狀態管理。Controller作為Kafka集群的核心組件，不僅負責分區的Leader選舉，還承擔著元數據的管理和同步任務。在實際生產環境中，Controller的可靠性和性能直接影響到整個Kafka集群的可用性。因此，理解其底層源碼對我們優化和維護Kafka集群至關重要。
