Apache Kafka Controller 是 Kafka 集群的核心协调组件,主要承担两大核心职责:
一、核心作用
- 分区领导者选举 1
// 分区领导者选举逻辑示例(伪代码)
def electLeader(partition: Partition): Unit = {val isr = partition.inSyncReplicasval newLeader = isr.minBy(_.brokerId) // 实际使用更复杂算法partition.leader = newLeader
}
- 副本状态管理
- 维护所有分区的 ISR(In-Sync Replicas)列表
- 监控副本同步状态(通过 replica.lag.time.max.ms 参数)
- 集群元数据维护
- 处理 Broker 上下线事件
- 触发分区重分配(如使用 kafka-reassign-partitions 时)
二、故障恢复机制
- Controller 选举流程 2
# ZooKeeper 临时节点路径
ls /controller # 查看当前 Controller ID# 选举过程伪代码
1. 监控 /controller 节点消失
2. 各 Broker 尝试创建临时节点
3. 最先创建成功的成为新 Controller
- 故障恢复步骤:
- 元数据重建:从 ZooKeeper 加载 topics/brokers 数据
- 状态机重置:重建分区状态机(PartitionStateMachine)和副本状态机(ReplicaStateMachine)
- 补偿操作:处理故障期间未完成的管控请求
补充说明:
- 从 Kafka 2.8 版本开始支持 KRaft 模式(不依赖 ZooKeeper),故障恢复机制改为基于 Raft 共识算法 3
- Controller 故障期间,已有生产消费不受影响,但无法执行管理操作(如创建 topic)