Kafka 源码解析：集群协同运行控制器

class KafkaController(val config: KafkaConfig, // 配置信息
                      zkUtils: ZkUtils, // ZK 交互 工具 类
                      val brokerState: BrokerState, // 描述 broker 节点的状态
                      time: Time, // 时间戳工具类
                      metrics: Metrics,
                      threadNamePrefix: Option[String] = None) extends Logging with KafkaMetricsGroup {

    /** 标识是否启动 */
    private var isRunning = true
    /** 维护上下文信息，缓存 ZK 中记录的整个集群的元数据信息 */
    val controllerContext = new ControllerContext(zkUtils)
    /** 管理集群中所有分区状态的状态机 */
    val partitionStateMachine = new PartitionStateMachine(this)
    /** 管理集群中所有副本状态的状态机 */
    val replicaStateMachine = new ReplicaStateMachine(this)
    /** 用于故障转移，选举 leader */
    private val controllerElector = new ZookeeperLeaderElector(
        controllerContext, ZkUtils.ControllerPath, onControllerFailover, onControllerResignation, config.brokerId, time)
    /** 分区再平衡定时任务调度器 */
    private val autoRebalanceScheduler = new KafkaScheduler(1)
    /** 用于对指定的 topic 执行删除操作 */
    var deleteTopicManager: TopicDeletionManager = _
    /** Leader 副本选举策略 */
    val offlinePartitionSelector = new OfflinePartitionLeaderSelector(controllerContext, config)
    private val reassignedPartitionLeaderSelector = new ReassignedPartitionLeaderSelector(controllerContext)
    private val preferredReplicaPartitionLeaderSelector = new PreferredReplicaPartitionLeaderSelector(controllerContext)
    private val controlledShutdownPartitionLeaderSelector = new ControlledShutdownLeaderSelector(controllerContext)
    /** 用于批量向 broker 节点发送请求 */
    private val brokerRequestBatch = new ControllerBrokerRequestBatch(this)
    /** ZK 监听器 */
    private val partitionReassignedListener = new PartitionsReassignedListener(this)
    private val preferredReplicaElectionListener = new PreferredReplicaElectionListener(this)
    private val isrChangeNotificationListener = new IsrChangeNotificationListener(this)

    // ... 省略方法定义

}

def startup(): Unit = {
    inLock(controllerContext.controllerLock) {
        info("Controller starting up")
        // 注册 SessionExpirationListener 监听器，监听 Controller 与 ZK 的连接状态
        this.registerSessionExpirationListener()
        // 标识当前 controller 已经启动，现在还是 follower 角色
        isRunning = true
        // 启动故障转移机制
        controllerElector.startup
        info("Controller startup complete")
    }
}

class ControllerContext(val zkUtils: ZkUtils) {

    /** 管理 controller 与集群中其它 broker 之间的连接 */
    var controllerChannelManager: ControllerChannelManager = _
    /** 记录正在关闭的 brokerId 集合 */
    var shuttingDownBrokerIds: mutable.Set[Int] = mutable.Set.empty
    /** controller 的年代信息，初始为 0，每次重新选举之后值加 1 */
    var epoch: Int = KafkaController.InitialControllerEpoch - 1
    /** 年代信息对应的 ZK 版本，初始为 0  */
    var epochZkVersion: Int = KafkaController.InitialControllerEpochZkVersion - 1
    /** 集群中全部的 topic 集合  */
    var allTopics: Set[String] = Set.empty
    /** 记录每个分区对应的 AR 集合 */
    var partitionReplicaAssignment: mutable.Map[TopicAndPartition, Seq[Int]] = mutable.Map.empty
    /** 记录每个分区的 leader 副本所在的 brokerId、ISR 集合，以及 controller 年代信息 */
    var partitionLeadershipInfo: mutable.Map[TopicAndPartition, LeaderIsrAndControllerEpoch] = mutable.Map.empty
    /** 记录正在重新分配副本的分区 */
    val partitionsBeingReassigned: mutable.Map[TopicAndPartition, ReassignedPartitionsContext] = new mutable.HashMap
    /** 记录了正在进行优先副本选举的分区 */
    val partitionsUndergoingPreferredReplicaElection: mutable.Set[TopicAndPartition] = new mutable.HashSet
    /** 记录了当前可用的 broker 集合 */
    private var liveBrokersUnderlying: Set[Broker] = Set.empty
    /** 记录了当前可用的 brokerId 集合 */
    private var liveBrokerIdsUnderlying: Set[Int] = Set.empty

    // ... 省略方法定义

}

case class ControllerBrokerStateInfo(networkClient: NetworkClient,
                                     brokerNode: Node,
                                     messageQueue: BlockingQueue[QueueItem],
                                     requestSendThread: RequestSendThread)

private def addNewBroker(broker: Broker) {
    // 创建消息队列，用于存放发往指定 broker 节点的请求
    val messageQueue = new LinkedBlockingQueue[QueueItem]
    debug("Controller %d trying to connect to broker %d".format(config.brokerId, broker.id))
    val brokerEndPoint = broker.getBrokerEndPoint(config.interBrokerListenerName)
    val brokerNode = new Node(broker.id, brokerEndPoint.host, brokerEndPoint.port)
    // 创建网络连接客户端
    val networkClient = {
        val channelBuilder = ChannelBuilders.clientChannelBuilder(
            config.interBrokerSecurityProtocol,
            LoginType.SERVER,
            config.values,
            config.saslMechanismInterBrokerProtocol,
            config.saslInterBrokerHandshakeRequestEnable
        )
        val selector = new Selector(
            NetworkReceive.UNLIMITED,
            Selector.NO_IDLE_TIMEOUT_MS,
            metrics,
            time,
            "controller-channel",
            Map("broker-id" -> broker.id.toString).asJava,
            false,
            channelBuilder
        )
        new NetworkClient(
            selector,
            new ManualMetadataUpdater(Seq(brokerNode).asJava),
            config.brokerId.toString,
            1,
            0,
            Selectable.USE_DEFAULT_BUFFER_SIZE,
            Selectable.USE_DEFAULT_BUFFER_SIZE,
            config.requestTimeoutMs,
            time,
            false
        )
    }

    // 创建请求发送线程对象
    val threadName = threadNamePrefix match {
        case None => "Controller-%d-to-broker-%d-send-thread".format(config.brokerId, broker.id)
        case Some(name) => "%s:Controller-%d-to-broker-%d-send-thread".format(name, config.brokerId, broker.id)
    }
    val requestThread = new RequestSendThread(
        config.brokerId, controllerContext, messageQueue, networkClient, brokerNode, config, time, threadName)
    requestThread.setDaemon(false)

    // 记录到 brokerStateInfo 集合
    brokerStateInfo.put(broker.id, ControllerBrokerStateInfo(networkClient, brokerNode, messageQueue, requestThread))
}

override def doWork(): Unit = {

    def backoff(): Unit = CoreUtils.swallowTrace(Thread.sleep(100))

    // 获取缓冲队列中的 QueueItem 对象，封装了请求类型、请求对象，以及响应回调函数
    val QueueItem(apiKey, requestBuilder, callback) = queue.take()

    import NetworkClientBlockingOps._ // 阻塞模式

    var clientResponse: ClientResponse = null
    try {
        lock synchronized {
            var isSendSuccessful = false // 标识请求是否发送成功
            while (isRunning.get() && !isSendSuccessful) {
                // 当 broker 节点宕机后，会触发 ZK 的监听器调用 removeBroker 方法停止当前线程，在停止前会一直尝试重试
                try {
                    // 阻塞等待指定 broker 节点是否允许接收请求
                    if (!brokerReady()) {
                        isSendSuccessful = false
                        backoff() // 等待 100 毫秒
                    } else {
                        // 创建并发送请求，同时阻塞等待响应
                        val clientRequest = networkClient.newClientRequest(brokerNode.idString, requestBuilder, time.milliseconds(), true)
                        clientResponse = networkClient.blockingSendAndReceive(clientRequest)(time)
                        isSendSuccessful = true
                    }
                } catch {
                    // ... 省略异常处理，如果发送失败，则断开连接，并等待一段时间后重试
                }
            }

            // 解析响应
            if (clientResponse != null) {
                // 解析请求类型
                val api = ApiKeys.forId(clientResponse.requestHeader.apiKey)
                // 只允许 LEADER_AND_ISR、STOP_REPLICA，以及 UPDATE_METADATA_KEY 这 3 种请求类型
                if (api != ApiKeys.LEADER_AND_ISR && api != ApiKeys.STOP_REPLICA && api != ApiKeys.UPDATE_METADATA_KEY)
                    throw new KafkaException(s"Unexpected apiKey received: $apiKey")

                // 执行响应回调函数
                val response = clientResponse.responseBody
                if (callback != null) {
                    callback(response)
                }
            }
        }
    } catch {
        // ... 省略异常处理
    }
}

class ControllerBrokerRequestBatch(controller: KafkaController) extends Logging {

    /** Controller 上下文信息 */
    val controllerContext: ControllerContext = controller.controllerContext
    /** Controller 节点 ID */
    val controllerId: Int = controller.config.brokerId
    /** 记录发往指定 broker 节点的 LeaderAndIsrRequest 请求所需的信息 */
    val leaderAndIsrRequestMap = mutable.Map.empty[Int, mutable.Map[TopicPartition, PartitionStateInfo]]
    /** 记录发往指定 broker 节点的 StopReplicaRequest 请求所需的信息 */
    val stopReplicaRequestMap = mutable.Map.empty[Int, Seq[StopReplicaRequestInfo]]
    /** UpdateMetadataRequest 请求对应的目标 brokerId 集合 */
    val updateMetadataRequestBrokerSet = mutable.Set.empty[Int]
    /** UpdateMetadataRequest 请求对应的请求信息 */
    val updateMetadataRequestPartitionInfoMap = mutable.Map.empty[TopicPartition, PartitionStateInfo]

    // ... 省略方法定义

}

class PartitionStateMachine(controller: KafkaController) extends Logging {

    /** Controller 的上下文信息 */
    private val controllerContext = controller.controllerContext
    /** Controller 节点的 ID */
    private val controllerId = controller.config.brokerId
    /** ZK 工具类 */
    private val zkUtils = controllerContext.zkUtils
    /** 记录每个分区对应的分区状态信息 */
    private val partitionState: mutable.Map[TopicAndPartition, PartitionState] = mutable.Map.empty
    /** 用于向指定的 broker 批量发送请求 */
    private val brokerRequestBatch = new ControllerBrokerRequestBatch(controller)
    /** 表示分区状态机是否已经启动 */
    private val hasStarted = new AtomicBoolean(false)
    /** 默认 leader 副本选举器 */
    private val noOpPartitionLeaderSelector = new NoOpLeaderSelector(controllerContext)
    /** 用于监听 topic 变化的 ZK 监听器 */
    private val topicChangeListener = new TopicChangeListener(controller)
    /** 用于监听 topic 删除的 ZK 监听器 */
    private val deleteTopicsListener = new DeleteTopicsListener(controller)
    /** 记录监听对应 topic 分区变化的监听器集合 */
    private val partitionModificationsListeners: mutable.Map[String, PartitionModificationsListener] = mutable.Map.empty

    // ... 省略方法定义

}

def startup() {
    // 初始化本地记录的所有分区状态
    this.initializePartitionState()
    // 标识分区状态机已经启动
    hasStarted.set(true)
    // 尝试将集群中所有 OfflinePartition 或 NewPartition 状态的可用分区切换成 OnlinePartition 状态
    this.triggerOnlinePartitionStateChange()
}

private def initializePartitionState() {
    // 遍历集群中的所有分区
    for (topicPartition <- controllerContext.partitionReplicaAssignment.keys) {
        // 获取对应分区 leader 副本所在的 brokerId、ISR 集合，以及 controller 年代信息
        controllerContext.partitionLeadershipInfo.get(topicPartition) match {
            // 存在 leader 副本和 ISR 集合
            case Some(currentLeaderIsrAndEpoch) =>
                // 分区 leader 副本所在的 broker 可用，初始化分区为 OnlinePartition 状态
                if (controllerContext.liveBrokerIds.contains(currentLeaderIsrAndEpoch.leaderAndIsr.leader))
                    partitionState.put(topicPartition, OnlinePartition)
                // 分区 leader 副本所在的 broker 不可用，初始化为 OfflinePartition 状态
                else
                    partitionState.put(topicPartition, OfflinePartition)
            // 如果不存在，则说明是一个新创建的分区，设置分区状态为 NewPartition
            case None =>
                partitionState.put(topicPartition, NewPartition)
        }
    }
}

def triggerOnlinePartitionStateChange() {
    try {
        // 校验待发送的请求集合，确保历史的请求已经全部发送完毕
        brokerRequestBatch.newBatch()
        // 遍历处理集群中所有的分区，不包含正在等待被删除的 topic 的分区，尝试切换分区状态为 OnlinePartition
        for ((topicAndPartition, partitionState) <- partitionState
             if !controller.deleteTopicManager.isTopicQueuedUpForDeletion(topicAndPartition.topic)) {
            // 对于 OfflinePartition 或 NewPartition 状态的分区，尝试将对应分区状态修改为 OnlinePartition 状态
            if (partitionState.equals(OfflinePartition) || partitionState.equals(NewPartition))
                this.handleStateChange(
                    topicAndPartition.topic,
                    topicAndPartition.partition,
                    OnlinePartition,
                    controller.offlinePartitionSelector,
                    (new CallbackBuilder).build)
        }
        // 发送请求
        brokerRequestBatch.sendRequestsToBrokers(controller.epoch)
    } catch {
        // ... 省略异常处理
    }
}

def handleStateChanges(partitions: Set[TopicAndPartition], // 待处理的 topic 分区集合
                       targetState: PartitionState, // 目标分区状态
                       leaderSelector: PartitionLeaderSelector = noOpPartitionLeaderSelector, // 分区 leader 副本选择器
                       callbacks: Callbacks = (new CallbackBuilder).build) {
    info("Invoking state change to %s for partitions %s".format(targetState, partitions.mkString(",")))
    try {
        // 校验待发送的请求集合，确保历史的请求已经全部发送完毕
        brokerRequestBatch.newBatch()
        // 遍历待处理的 topic 分区集合，执行分区状态切换
        partitions.foreach { topicAndPartition =>
            this.handleStateChange(topicAndPartition.topic, topicAndPartition.partition, targetState, leaderSelector, callbacks)
        }
        // 发送请求
        brokerRequestBatch.sendRequestsToBrokers(controller.epoch)
    } catch {
        // ... 省略异常处理
    }
}

private def handleStateChange(topic: String, partition: Int,
                              targetState: PartitionState,
                              leaderSelector: PartitionLeaderSelector, // 执行 leader 选举的选择器
                              callbacks: Callbacks) {
    val topicAndPartition = TopicAndPartition(topic, partition)

    // 检测当前分区状态机是否已经启动，只有 kafka controller leader 的分区状态机才需要启动，如果没有启动则抛出异常
    if (!hasStarted.get)
        throw new StateChangeFailedException(("Controller %d epoch %d initiated state change for partition %s to %s failed because " +
                "the partition state machine has not started").format(controllerId, controller.epoch, topicAndPartition, targetState))

    // 获取分区的当前状态，没有则初始化为 NonExistentPartition 状态
    val currState = partitionState.getOrElseUpdate(topicAndPartition, NonExistentPartition)
    try {
        // 在转换开始之前，会依据于 targetState 检查分区的前置状态是否合法
        targetState match {
            case NewPartition =>
                // 如果目标状态为 NewPartition，则前置状态必须是 NonExistentPartition
                this.assertValidPreviousStates(topicAndPartition, List(NonExistentPartition), NewPartition)
                // 切换分区状态为 NewPartition
                partitionState.put(topicAndPartition, NewPartition)
            case OnlinePartition =>
                // 如果目标状态为 OnlinePartition，则前置状态必须是 NewPartition, OnlinePartition, OfflinePartition 中的一个
                this.assertValidPreviousStates(topicAndPartition, List(NewPartition, OnlinePartition, OfflinePartition), OnlinePartition)
                partitionState(topicAndPartition) match {
                    case NewPartition =>
                        // 如果前置状态是 NewPartition，则需要为分区分配 leader 副本和 ISR 集合
                        this.initializeLeaderAndIsrForPartition(topicAndPartition)
                    case OfflinePartition =>
                        // 如果前置状态是 OfflinePartition，则需要为分区选举新的 leader 副本
                        this.electLeaderForPartition(topic, partition, leaderSelector)
                    case OnlinePartition => // invoked when the leader needs to be re-elected
                        // 如果前置状态为 OnlinePartition，则需要为分区重新选举新的 leader 副本
                        this.electLeaderForPartition(topic, partition, leaderSelector)
                    case _ => // should never come here since illegal previous states are checked above
                }
                // 设置分区状态为 OnlinePartition
                partitionState.put(topicAndPartition, OnlinePartition)
            case OfflinePartition =>
                // 如果目标状态为 OfflinePartition，则前置状态必须是 NewPartition, OnlinePartition, OfflinePartition 中的一个
                this.assertValidPreviousStates(topicAndPartition, List(NewPartition, OnlinePartition, OfflinePartition), OfflinePartition)
                // 设置分区状态为 OfflinePartition
                partitionState.put(topicAndPartition, OfflinePartition)
            case NonExistentPartition =>
                // 如果目标状态为 NonExistentPartition，则前置状态必须是 OfflinePartition
                this.assertValidPreviousStates(topicAndPartition, List(OfflinePartition), NonExistentPartition)
                // 设置分区状态为 NonExistentPartition
                partitionState.put(topicAndPartition, NonExistentPartition)
        }
    } catch {
        // ... 省略异常处理
    }
}

private def initializeLeaderAndIsrForPartition(topicAndPartition: TopicAndPartition) {
    // 获取分区的 AR 集合
    val replicaAssignment = controllerContext.partitionReplicaAssignment(topicAndPartition)
    // 获取 AR 集合中可用的副本集合
    val liveAssignedReplicas = replicaAssignment.filter(r => controllerContext.liveBrokerIds.contains(r))
    liveAssignedReplicas.size match {
        // 没有可用的副本，抛出异常
        case 0 =>
            val failMsg = "encountered error during state change of partition %s from New to Online, assigned replicas are [%s], live brokers are [%s]. No assigned replica is alive."
                    .format(topicAndPartition, replicaAssignment.mkString(","), controllerContext.liveBrokerIds)
            stateChangeLogger.error("Controller %d epoch %d ".format(controllerId, controller.epoch) + failMsg)
            throw new StateChangeFailedException(failMsg)
        case _ =>
            debug("Live assigned replicas for partition %s are: [%s]".format(topicAndPartition, liveAssignedReplicas))
            // 将可用的 AR 集合中的第一个副本选为 leader 副本
            val leader = liveAssignedReplicas.head
            // 创建 LeaderIsrAndControllerEpoch 对象，其中的 ISR 集合是可用的 AR 集合
            val leaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(new LeaderAndIsr(leader, liveAssignedReplicas.toList), controller.epoch)
            debug("Initializing leader and isr for partition %s to %s".format(topicAndPartition, leaderIsrAndControllerEpoch))
            try {
                // 将分区 leader 副本和 ISR 集合等信息写入 ZK，路径：/brokers/topics/{topic_name}/partitions/{partitionId}/state
                zkUtils.createPersistentPath(
                    getTopicPartitionLeaderAndIsrPath(topicAndPartition.topic, topicAndPartition.partition),
                    zkUtils.leaderAndIsrZkData(leaderIsrAndControllerEpoch.leaderAndIsr, controller.epoch))
                // 更新本地缓存的指定 topic 分区的相关信息
                controllerContext.partitionLeadershipInfo.put(topicAndPartition, leaderIsrAndControllerEpoch)
                // 添加 LeaderAndIsrRequest 请求，待发送
                brokerRequestBatch.addLeaderAndIsrRequestForBrokers(
                    liveAssignedReplicas, topicAndPartition.topic, topicAndPartition.partition, leaderIsrAndControllerEpoch, replicaAssignment)
            } catch {
                // ... 省略异常处理
            }
    }
}

def electLeaderForPartition(topic: String, partition: Int, leaderSelector: PartitionLeaderSelector) {
    val topicAndPartition = TopicAndPartition(topic, partition)
    try {
        var zookeeperPathUpdateSucceeded: Boolean = false
        var newLeaderAndIsr: LeaderAndIsr = null
        var replicasForThisPartition: Seq[Int] = Seq.empty[Int]
        while (!zookeeperPathUpdateSucceeded) {
            // 从 ZK 获取分区当前的 leader 副本、ISR 集合、zkVersion 等信息，如果不存在则抛出异常
            val currentLeaderIsrAndEpoch = this.getLeaderIsrAndEpochOrThrowException(topic, partition)
            val currentLeaderAndIsr = currentLeaderIsrAndEpoch.leaderAndIsr
            val controllerEpoch = currentLeaderIsrAndEpoch.controllerEpoch
            // 检测 controller 的年代信息，如果当前年代信息小于 ZK 中记录的年代信息，则说明存在新的 controller，需要放弃本次选举操作
            if (controllerEpoch > controller.epoch) {
                // ... 抛出 StateChangeFailedException 异常，省略
            }
            // 使用指定的 leader 副本选举器选择新的 leader 副本和 ISR 集合
            val (leaderAndIsr, replicas) = leaderSelector.selectLeader(topicAndPartition, currentLeaderAndIsr)
            // 将新的 leader 副本和 ISR 集合信息转换成 JSON 格式记录到 ZK，路径：/brokers/topics/{topic_name}/partitions/{partitionId}/state
            val (updateSucceeded, newVersion) = ReplicationUtils.updateLeaderAndIsr(
                zkUtils, topic, partition, leaderAndIsr, controller.epoch, currentLeaderAndIsr.zkVersion)
            newLeaderAndIsr = leaderAndIsr
            newLeaderAndIsr.zkVersion = newVersion
            zookeeperPathUpdateSucceeded = updateSucceeded
            replicasForThisPartition = replicas
        }
        val newLeaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(newLeaderAndIsr, controller.epoch)
        // 更新本地缓存的指定 topic 分区的相关信息
        controllerContext.partitionLeadershipInfo.put(TopicAndPartition(topic, partition), newLeaderIsrAndControllerEpoch)
        // 获取指定分区的 AR 集合
        val replicas = controllerContext.partitionReplicaAssignment(TopicAndPartition(topic, partition))
        // 添加 LeaderAndIsrRequest 请求，待发送
        brokerRequestBatch.addLeaderAndIsrRequestForBrokers(replicasForThisPartition, topic, partition, newLeaderIsrAndControllerEpoch, replicas)
    } catch {
        // ... 省略异常处理
    }
    debug("After leader election, leader cache is updated to %s".format(controllerContext.partitionLeadershipInfo.map(l => (l._1, l._2))))
}

trait PartitionLeaderSelector {

    /**
     * 选举分区 leader 副本
     *
     * @param topicAndPartition   需要执行 leader 副本选举的分区
     * @param currentLeaderAndIsr 目标分区当前 leader 副本、ISR 集合
     * @return 选举后的新的 leader 副本和新 ISR 集合信息，以及需要接收 LeaderAndIsrRequest 的 brokerId 集合
     * @throws NoReplicaOnlineException 如果 AR 集合中不存在可用的副本，则抛出异常
     */
    def selectLeader(topicAndPartition: TopicAndPartition, currentLeaderAndIsr: LeaderAndIsr): (LeaderAndIsr, Seq[Int])

}

def selectLeader(topicAndPartition: TopicAndPartition, currentLeaderAndIsr: LeaderAndIsr): (LeaderAndIsr, Seq[Int]) = {
    controllerContext.partitionReplicaAssignment.get(topicAndPartition) match {
        // 处理分区的 AR 集合
        case Some(assignedReplicas) =>
            // 获取分区 AR 集合中可用的副本
            val liveAssignedReplicas = assignedReplicas.filter(r => controllerContext.liveBrokerIds.contains(r))
            // 获取 ISR 集合中可用的副本
            val liveBrokersInIsr = currentLeaderAndIsr.isr.filter(r => controllerContext.liveBrokerIds.contains(r))
            val currentLeaderEpoch = currentLeaderAndIsr.leaderEpoch
            val currentLeaderIsrZkPathVersion = currentLeaderAndIsr.zkVersion

            // 依据 ISR 集合中是否有可用的副本决定是从 ISR 集合中选举新的 leader 副本，还是从 AR 集合中选举新的 leader 副本
            val newLeaderAndIsr =
                if (liveBrokersInIsr.isEmpty) { // 如果 ISR 集合中不存在可用的副本
                    // 依据配置（unclean.leader.election.enable）决定是否从 AR 集合中选择 leader 副本，如果不允许则抛出异常
                    if (!LogConfig.fromProps(config.originals, AdminUtils.fetchEntityConfig(
                        controllerContext.zkUtils, ConfigType.Topic, topicAndPartition.topic)).uncleanLeaderElectionEnable) {
                        // ... 抛出 NoReplicaOnlineException 异常，省略
                    }
                    debug("No broker in ISR is alive for %s. Pick the leader from the alive assigned replicas: %s".format(topicAndPartition, liveAssignedReplicas.mkString(",")))
                    if (liveAssignedReplicas.isEmpty) {
                        // AR 集合中没有可用副本，直接抛出异常
                        // ... 抛出 NoReplicaOnlineException 异常，省略
                    } else {
                        // 从 AR 集合可用的副本中选取第一个副本作为新的 leader 副本，新的 ISR 集合中只有新的 leader 副本自己
                        ControllerStats.uncleanLeaderElectionRate.mark()
                        val newLeader = liveAssignedReplicas.head
                        warn("No broker in ISR is alive for %s. Elect leader %d from live brokers %s. There's potential data loss.".format(topicAndPartition, newLeader, liveAssignedReplicas.mkString(",")))
                        new LeaderAndIsr(newLeader, currentLeaderEpoch + 1, List(newLeader), currentLeaderIsrZkPathVersion + 1)
                    }
                } else { // ISR 集合中存在可用的副本
                    val liveReplicasInIsr = liveAssignedReplicas.filter(r => liveBrokersInIsr.contains(r))
                    // 从 ISR 集合中选择第一个副本作为新的 leader 副本，以 ISR 集合中可用的副本集合作为新的 ISR 集合
                    val newLeader = liveReplicasInIsr.head
                    debug("Some broker in ISR is alive for %s. Select %d from ISR %s to be the leader.".format(topicAndPartition, newLeader, liveBrokersInIsr.mkString(",")))
                    // 构造 LeaderAndIsr 对象并返回
                    new LeaderAndIsr(newLeader, currentLeaderEpoch + 1, liveBrokersInIsr, currentLeaderIsrZkPathVersion + 1)
                }
            info("Selected new leader and ISR %s for offline partition %s".format(newLeaderAndIsr.toString(), topicAndPartition))

            // 需要向 AR 集合中所有的可用副本所在 broker 节点发送 LeaderAndIsrRequest 请求
            (newLeaderAndIsr, liveAssignedReplicas)
        // 当前 topic 分区没有可用的副本
        case None =>
            throw new NoReplicaOnlineException("Partition %s doesn't have replicas assigned to it".format(topicAndPartition))
    }
}

def selectLeader(topicAndPartition: TopicAndPartition, currentLeaderAndIsr: LeaderAndIsr): (LeaderAndIsr, Seq[Int]) = {
    // 获取新分配的 AR 集合
    val reassignedInSyncReplicas = controllerContext.partitionsBeingReassigned(topicAndPartition).newReplicas
    val currentLeaderEpoch = currentLeaderAndIsr.leaderEpoch
    val currentLeaderIsrZkPathVersion = currentLeaderAndIsr.zkVersion
    // 新选择的 leader 副本必须在新分配的 AR 集合和 ISR 集合中
    val aliveReassignedInSyncReplicas = reassignedInSyncReplicas
            .filter(r => controllerContext.liveBrokerIds.contains(r) && currentLeaderAndIsr.isr.contains(r))
    val newLeaderOpt = aliveReassignedInSyncReplicas.headOption
    newLeaderOpt match {
        // 存在满足条件的 leader 副本，以之前的 ISR 集合作为新的 ISR 集合
        case Some(newLeader) =>
            (new LeaderAndIsr(newLeader, currentLeaderEpoch + 1, currentLeaderAndIsr.isr, currentLeaderIsrZkPathVersion + 1), reassignedInSyncReplicas)
        // 不存在满足条件的 leader 副本，抛出异常
        case None =>
            reassignedInSyncReplicas.size match {
                case 0 =>
                    throw new NoReplicaOnlineException("List of reassigned replicas for partition %s is empty. Current leader and ISR: [%s]".format(topicAndPartition, currentLeaderAndIsr))
                case _ =>
                    throw new NoReplicaOnlineException("None of the reassigned replicas for partition %s are in-sync with the leader. Current leader and ISR: [%s]".format(topicAndPartition, currentLeaderAndIsr))
            }
    }
}

def selectLeader(topicAndPartition: TopicAndPartition, currentLeaderAndIsr: LeaderAndIsr): (LeaderAndIsr, Seq[Int]) = {
    // 获取分区 AR 集合
    val assignedReplicas = controllerContext.partitionReplicaAssignment(topicAndPartition)
    // 以分区 AR 集合的第一个副本作为优先副本
    val preferredReplica = assignedReplicas.head
    val currentLeader = controllerContext.partitionLeadershipInfo(topicAndPartition).leaderAndIsr.leader
    if (currentLeader == preferredReplica) {
        // 优先副本已经是 leader 副本
        throw new LeaderElectionNotNeededException("Preferred replica %d is already the current leader for partition %s".format(preferredReplica, topicAndPartition))
    } else {
        info("Current leader %d for partition %s is not the preferred replica.".format(currentLeader, topicAndPartition) + " Triggering preferred replica leader election")
        // 如果优先副本所在的 broker 节点可用，且位于 ISR 集合中，则选择成为新的 leader 副本
        if (controllerContext.liveBrokerIds.contains(preferredReplica) && currentLeaderAndIsr.isr.contains(preferredReplica)) {
            (new LeaderAndIsr(preferredReplica, currentLeaderAndIsr.leaderEpoch + 1, currentLeaderAndIsr.isr, currentLeaderAndIsr.zkVersion + 1), assignedReplicas)
        } else {
            throw new StateChangeFailedException("Preferred replica %d for partition ".format(preferredReplica) + "%s is either not alive or not in the isr. Current leader and ISR: [%s]".format(topicAndPartition, currentLeaderAndIsr))
        }
    }
}

def selectLeader(topicAndPartition: TopicAndPartition, currentLeaderAndIsr: LeaderAndIsr): (LeaderAndIsr, Seq[Int]) = {
    val currentLeaderEpoch = currentLeaderAndIsr.leaderEpoch
    val currentLeaderIsrZkPathVersion = currentLeaderAndIsr.zkVersion
    val currentLeader = currentLeaderAndIsr.leader
    val assignedReplicas = controllerContext.partitionReplicaAssignment(topicAndPartition)
    // 获取当前可用的 brokerId 集合
    val liveOrShuttingDownBrokerIds = controllerContext.liveOrShuttingDownBrokerIds
    val liveAssignedReplicas = assignedReplicas.filter(r => liveOrShuttingDownBrokerIds.contains(r))
    // 从当前 ISR 集合中移除副本所在 broker 节点正在关闭的副本
    val newIsr = currentLeaderAndIsr.isr.filter(brokerId => !controllerContext.shuttingDownBrokerIds.contains(brokerId))
    // 从可用的 AR 集合中选择位于新的 ISR 集合中的副本作为 leader 副本
    liveAssignedReplicas.find(newIsr.contains) match {
        case Some(newLeader) =>
            debug("Partition %s : current leader = %d, new leader = %d".format(topicAndPartition, currentLeader, newLeader))
            (LeaderAndIsr(newLeader, currentLeaderEpoch + 1, newIsr, currentLeaderIsrZkPathVersion + 1), liveAssignedReplicas)
        case None =>
            throw new StateChangeFailedException("No other replicas in ISR %s for %s besides shutting down brokers %s".format(currentLeaderAndIsr.isr.mkString(","), topicAndPartition, controllerContext.shuttingDownBrokerIds.mkString(",")))
    }
}

class ReplicaStateMachine(controller: KafkaController) extends Logging {

    /** Controller 上下文信息 */
    private val controllerContext = controller.controllerContext
    /** Controller 的 ID */
    private val controllerId = controller.config.brokerId
    /** ZK 工具类 */
    private val zkUtils = controllerContext.zkUtils
    /** 记录每个副本对应的状态 */
    private val replicaState: mutable.Map[PartitionAndReplica, ReplicaState] = mutable.Map.empty
    /** ZK 监听器，用于监听 broker 节点的上下线 */
    private val brokerChangeListener = new BrokerChangeListener(controller)
    /** 用于向 broker 节点批量发送请求 */
    private val brokerRequestBatch = new ControllerBrokerRequestBatch(controller)
    /** 标识当前副本状态机是否成功启动 */
    private val hasStarted = new AtomicBoolean(false)

    // ... 省略方法定义

}

def startup() {
    // 初始化每个 topic 分区 AR 集合中的副本状态
    this.initializeReplicaState()
    // 标识当前副本状态机启动成功
    hasStarted.set(true)
    // 尝试将所有可用副本切换成 OnlineReplica 状态
    this.handleStateChanges(controllerContext.allLiveReplicas(), OnlineReplica)
}

private def initializeReplicaState() {
    for ((topicPartition, assignedReplicas) <- controllerContext.partitionReplicaAssignment) {
        val topic = topicPartition.topic
        val partition = topicPartition.partition
        // 遍历每个分区的 AR 集合，初始化对应分区的状态
        assignedReplicas.foreach { replicaId =>
            val partitionAndReplica = PartitionAndReplica(topic, partition, replicaId)
            // 将可用的副本初始化为 OnlineReplica 状态
            if (controllerContext.liveBrokerIds.contains(replicaId)) replicaState.put(partitionAndReplica, OnlineReplica)
            // 将不可用的副本初始化为 ReplicaDeletionIneligible 状态
            else replicaState.put(partitionAndReplica, ReplicaDeletionIneligible)
        }
    }
}

def handleStateChanges(replicas: Set[PartitionAndReplica],
                       targetState: ReplicaState,
                       callbacks: Callbacks = (new CallbackBuilder).build) {
    if (replicas.nonEmpty) {
        info("Invoking state change to %s for replicas %s".format(targetState, replicas.mkString(",")))
        try {
            // 校验待发送的请求集合，确保历史的请求已经全部发送完毕
            brokerRequestBatch.newBatch()
            // 遍历切换每个副本的状态
            replicas.foreach(r => this.handleStateChange(r, targetState, callbacks))
            // 发送请求
            brokerRequestBatch.sendRequestsToBrokers(controller.epoch)
        } catch {
            case e: Throwable => error("Error while moving some replicas to %s state".format(targetState), e)
        }
    }
}

def handleStateChange(partitionAndReplica: PartitionAndReplica, targetState: ReplicaState, callbacks: Callbacks) {
    val topic = partitionAndReplica.topic
    val partition = partitionAndReplica.partition
    val replicaId = partitionAndReplica.replica
    val topicAndPartition = TopicAndPartition(topic, partition)
    // 检测副本状态机是否已经启动
    if (!hasStarted.get)
        throw new StateChangeFailedException(
            "Controller %d epoch %d initiated state change of replica %d for partition %s to %s failed because replica state machine has not started"
                    .format(controllerId, controller.epoch, replicaId, topicAndPartition, targetState))

    // 获取指定副本当前的状态，如果不存在则初始化为 NonExistentReplica 状态
    val currState = replicaState.getOrElseUpdate(partitionAndReplica, NonExistentReplica)
    try {
        // 获取分区的 AR 集合
        val replicaAssignment = controllerContext.partitionReplicaAssignment(topicAndPartition)
        // 在转换开始之前，会依据于目标状态检查副本的前置状态是否合法
        targetState match {
            case NewReplica =>
                // 如果目标状态为 NewReplica，则前置状态必须是 NonExistentReplica
                this.assertValidPreviousStates(partitionAndReplica, List(NonExistentReplica), targetState)
                // 从 ZK 获取分区的 leader 副本和 ISR 集合等信息
                val leaderIsrAndControllerEpochOpt = ReplicationUtils.getLeaderIsrAndEpochForPartition(zkUtils, topic, partition)
                leaderIsrAndControllerEpochOpt match {
                    case Some(leaderIsrAndControllerEpoch) =>
                        // NewReplica 状态的副本不能是 leader 角色
                        if (leaderIsrAndControllerEpoch.leaderAndIsr.leader == replicaId)
                            throw new StateChangeFailedException("Replica %d for partition %s cannot be moved to NewReplica"
                                    .format(replicaId, topicAndPartition) + "state as it is being requested to become leader")
                        // 向该副本所在 broker 节点发送 LeaderAndIsrRequest 请求，并发送 UpdateMetadataRequest 给所有可用的 broker 节点
                        brokerRequestBatch.addLeaderAndIsrRequestForBrokers(
                            List(replicaId), topic, partition, leaderIsrAndControllerEpoch, replicaAssignment)
                    case None => // new leader request will be sent to this replica when one gets elected
                }
                // 更新当前副本状态为 NewReplica
                replicaState.put(partitionAndReplica, NewReplica)
            case ReplicaDeletionStarted =>
                // 如果目标状态为 ReplicaDeletionStarted，则前置状态必须是 OfflineReplica
                this.assertValidPreviousStates(partitionAndReplica, List(OfflineReplica), targetState)
                // 更新当前副本状态为 ReplicaDeletionStarted
                replicaState.put(partitionAndReplica, ReplicaDeletionStarted)
                // 向该副本所在 broker 节点发送 StopReplicaRequest 请求
                brokerRequestBatch.addStopReplicaRequestForBrokers(
                    List(replicaId), topic, partition, deletePartition = true, callbacks.stopReplicaResponseCallback)
            case ReplicaDeletionIneligible =>
                // 如果目标状态为 ReplicaDeletionIneligible，则前置状态必须是 ReplicaDeletionStarted
                this.assertValidPreviousStates(partitionAndReplica, List(ReplicaDeletionStarted), targetState)
                // 更新当前副本状态为 ReplicaDeletionIneligible
                replicaState.put(partitionAndReplica, ReplicaDeletionIneligible)
            case ReplicaDeletionSuccessful =>
                // 如果目标状态为 ReplicaDeletionSuccessful，则前置状态必须是 ReplicaDeletionStarted
                this.assertValidPreviousStates(partitionAndReplica, List(ReplicaDeletionStarted), targetState)
                // 更新当前副本状态为 ReplicaDeletionSuccessful
                replicaState.put(partitionAndReplica, ReplicaDeletionSuccessful)
            case NonExistentReplica =>
                // 如果目标状态为 NonExistentReplica，则前置状态必须是 ReplicaDeletionSuccessful
                this.assertValidPreviousStates(partitionAndReplica, List(ReplicaDeletionSuccessful), targetState)
                // 从对应 topic 分区的 AR 集合中移除当前副本
                val currentAssignedReplicas = controllerContext.partitionReplicaAssignment(topicAndPartition)
                controllerContext.partitionReplicaAssignment.put(topicAndPartition, currentAssignedReplicas.filterNot(_ == replicaId))
                // 移除副本缓存在本地的状态
                replicaState.remove(partitionAndReplica)
            case OnlineReplica =>
                // 如果目标状态为 OnlineReplica，则前置状态必须是 NewReplica、OnlineReplica、OfflineReplica 和 ReplicaDeletionIneligible 中的一个
                this.assertValidPreviousStates(partitionAndReplica,
                    List(NewReplica, OnlineReplica, OfflineReplica, ReplicaDeletionIneligible), targetState)
                replicaState(partitionAndReplica) match {
                    case NewReplica =>
                        // 如果前置状态是 NewReplica，则尝试将当前副本添加到对应分区的 AR 集合中
                        val currentAssignedReplicas = controllerContext.partitionReplicaAssignment(topicAndPartition)
                        if (!currentAssignedReplicas.contains(replicaId))
                            controllerContext.partitionReplicaAssignment.put(topicAndPartition, currentAssignedReplicas :+ replicaId)
                    case _ =>
                        // 检测是否存在 leader 副本
                        controllerContext.partitionLeadershipInfo.get(topicAndPartition) match {
                            // 如果存在 leader 副本，则向该副本所在 broker 节点发送 LeaderAndIsrRequest 请求，并发送 UpdateMetadataRequest 给所有可用的 broker 节点
                            case Some(leaderIsrAndControllerEpoch) =>
                                brokerRequestBatch.addLeaderAndIsrRequestForBrokers(
                                    List(replicaId), topic, partition, leaderIsrAndControllerEpoch, replicaAssignment)
                                replicaState.put(partitionAndReplica, OnlineReplica)
                            // 不存在 leader 副本
                            case None => // that means the partition was never in OnlinePartition state, this means the broker never
                            // started a log for that partition and does not have a high watermark value for this partition
                        }
                }
                // 更新当前副本状态为 OnlineReplica
                replicaState.put(partitionAndReplica, OnlineReplica)
            case OfflineReplica =>
                // 如果目标状态为 OfflineReplica，则前置状态必须是 NewReplica、OnlineReplica、OfflineReplica 和 ReplicaDeletionIneligible 中的一个
                this.assertValidPreviousStates(partitionAndReplica,
                    List(NewReplica, OnlineReplica, OfflineReplica, ReplicaDeletionIneligible), targetState)
                // 向副本所在 broker 节点发送 StopReplicaRequest 请求
                brokerRequestBatch.addStopReplicaRequestForBrokers(List(replicaId), topic, partition, deletePartition = false)
                // As an optimization, the controller removes dead replicas from the ISR
                val leaderAndIsrIsEmpty: Boolean =
                    controllerContext.partitionLeadershipInfo.get(topicAndPartition) match {
                        case Some(_) =>
                            // 将当前副本从所在分区的 ISR 集合中移除
                            controller.removeReplicaFromIsr(topic, partition, replicaId) match {
                                case Some(updatedLeaderIsrAndControllerEpoch) =>
                                    // send the shrunk ISR state change request to all the remaining alive replicas of the partition.
                                    val currentAssignedReplicas = controllerContext.partitionReplicaAssignment(topicAndPartition)
                                    if (!controller.deleteTopicManager.isPartitionToBeDeleted(topicAndPartition)) {
                                        // 向除当前副本以外可用的 AR 集合副本发送 LeaderAndIsrRequest 请求，并向集群中所有可用的 broker 发送 UpdateMetadataRequest 请求
                                        brokerRequestBatch.addLeaderAndIsrRequestForBrokers(currentAssignedReplicas.filterNot(_ == replicaId),
                                            topic, partition, updatedLeaderIsrAndControllerEpoch, replicaAssignment)
                                    }
                                    // 更新当前副本状态为 OfflineReplica
                                    replicaState.put(partitionAndReplica, OfflineReplica)
                                    false
                                case None =>
                                    true
                            }
                        case None =>
                            true
                    }
                if (leaderAndIsrIsEmpty && !controller.deleteTopicManager.isPartitionToBeDeleted(topicAndPartition))
                    throw new StateChangeFailedException(
                        "Failed to change state of replica %d for partition %s since the leader and isr path in zookeeper is empty".format(replicaId, topicAndPartition))
        }
    }
    catch {
        // ... 省略异常处理
    }
}

class TopicDeletionManager(controller: KafkaController,
                           initialTopicsToBeDeleted: Set[String] = Set.empty,
                           initialTopicsIneligibleForDeletion: Set[String] = Set.empty) extends Logging {

    /** Controller 上下文 */
    val controllerContext: ControllerContext = controller.controllerContext
    /** 管理分区状态的状态机 */
    val partitionStateMachine: PartitionStateMachine = controller.partitionStateMachine
    /** 管理副本状态的状态机 */
    val replicaStateMachine: ReplicaStateMachine = controller.replicaStateMachine
    /** 条件对象，用于同步其他线程与 deleteTopicsThread 线程 */
    val deleteTopicsCond: Condition = deleteLock.newCondition()
    /** 标识 topic 删除操作是否开始 */
    val deleteTopicStateChanged: AtomicBoolean = new AtomicBoolean(false)
    /** 执行 topic 删除操作的线程 */
    var deleteTopicsThread: DeleteTopicsThread = _
    /** 是否允许删除 topic，对应 delete.topic.enable 配置  */
    val isDeleteTopicEnabled: lang.Boolean = controller.config.deleteTopicEnable
    /** 记录待删除的 topic 集合 */
    val topicsToBeDeleted: mutable.Set[String] = if (isDeleteTopicEnabled) {
        mutable.Set.empty[String] ++ initialTopicsToBeDeleted
    } else {
        // 如果配置不允许删除，则将对应的 topic 从 ZK 对应节点（/admin/delete_topics）下移除
        val zkUtils = controllerContext.zkUtils
        for (topic <- initialTopicsToBeDeleted) {
            val deleteTopicPath = getDeleteTopicPath(topic)
            info("Removing " + deleteTopicPath + " since delete topic is disabled")
            zkUtils.zkClient.delete(deleteTopicPath)
        }
        mutable.Set.empty[String]
    }
    /** 记录不可删除的 topic 集合 */
    val topicsIneligibleForDeletion: mutable.Set[String] = mutable.Set.empty[String] ++ (initialTopicsIneligibleForDeletion & topicsToBeDeleted)
    /** 记录待删除的分区集合 */
    val partitionsToBeDeleted: mutable.Set[TopicAndPartition] = topicsToBeDeleted.flatMap(controllerContext.partitionsForTopic)

    // ... 省略方法定义

}

def start() {
    if (isDeleteTopicEnabled) {
        deleteTopicsThread = new DeleteTopicsThread()
        // 表示 topic 删除操作开始运行
        if (topicsToBeDeleted.nonEmpty) deleteTopicStateChanged.set(true)
        // 启动后台删除线程
        deleteTopicsThread.start()
    }
}

override def doWork() {
    // 等待线程被唤醒
    awaitTopicDeletionNotification()

    if (!isRunning.get) return

    inLock(controllerContext.controllerLock) {
        // 获取并处理待删除的 topic 集合
        val topicsQueuedForDeletion = Set.empty[String] ++ topicsToBeDeleted
        topicsQueuedForDeletion.foreach { topic =>
            // 如果当前 topic 的所有副本都已经被成功删除
            if (controller.replicaStateMachine.areAllReplicasForTopicDeleted(topic)) {
                // 变更 topic 及其分区和副本的状态，并从 ZK 和 Controller 上下文中移除 topic 相关的信息
                completeDeleteTopic(topic)
                info("Deletion of topic %s successfully completed".format(topic))
            }
            // 如果当前 topic 存在未完成删除的副本
            else {
                // 如果任一副本处于 ReplicaDeletionStarted 状态，则等待
                if (controller.replicaStateMachine.isAtLeastOneReplicaInDeletionStartedState(topic)) {
                    // ... 暂时先跳过该 topic，这里省略日志打点
                }
                // 否则说明 topic 还未达到执行删除的条件，或者存在某个副本删除失败（对应 ReplicaDeletionIneligible 状态）
                else {
                    // 如果任一副本处于 ReplicaDeletionIneligible 状态，则将对应状态重置为 OfflineReplica 状态后重试
                    if (controller.replicaStateMachine.isAnyReplicaInState(topic, ReplicaDeletionIneligible)) {
                        markTopicForDeletionRetry(topic)
                    }
                }
            }

            // 检测当前 topic 是否可以删除
            if (isTopicEligibleForDeletion(topic)) {
                info("Deletion of topic %s (re)started".format(topic))
                // 开始执行 topic 删除操作
                onTopicDeletion(Set(topic))
            } else if (isTopicIneligibleForDeletion(topic)) {
                info("Not retrying deletion of topic %s at this time since it is marked ineligible for deletion".format(topic))
            }
        }
    }
}

private def completeDeleteTopic(topic: String) {
    // 1. 注销 PartitionModificationsListener 监听器
    partitionStateMachine.deregisterPartitionChangeListener(topic)

    // 2. 将指定 topic 下被成功删除的副本状态切换成 NonExistentReplica
    val replicasForDeletedTopic = controller.replicaStateMachine.replicasInState(topic, ReplicaDeletionSuccessful)
    replicaStateMachine.handleStateChanges(replicasForDeletedTopic, NonExistentReplica)

    // 3. 将指定 topic 下所有的 topic 分区状态切换成 NonExistentPartition
    val partitionsForDeletedTopic = controllerContext.partitionsForTopic(topic)
    partitionStateMachine.handleStateChanges(partitionsForDeletedTopic, OfflinePartition)
    partitionStateMachine.handleStateChanges(partitionsForDeletedTopic, NonExistentPartition)

    // 4. 将 topic 及其分区从待删除集合中移除
    topicsToBeDeleted -= topic
    partitionsToBeDeleted.retain(_.topic != topic)

    // 5. 清除 ZK 上与该 topic 相关的数据
    val zkUtils = controllerContext.zkUtils
    zkUtils.zkClient.deleteRecursive(getTopicPath(topic))
    zkUtils.zkClient.deleteRecursive(getEntityConfigPath(ConfigType.Topic, topic))
    zkUtils.zkClient.delete(getDeleteTopicPath(topic))

    // 6. 清空 Controller 上下文中与该 topic 相关的数据
    controllerContext.removeTopic(topic)
}

private def onTopicDeletion(topics: Set[String]) {
    info("Topic deletion callback for %s".format(topics.mkString(",")))
    val partitions = topics.flatMap(controllerContext.partitionsForTopic)
    // 向所有可用的 broker 节点发送 UpdateMetadataRequest 请求，通知当前 topic 需要被删除
    controller.sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq, partitions)
    // 按照 topic 进行分组
    val partitionReplicaAssignmentByTopic = controllerContext.partitionReplicaAssignment.groupBy(p => p._1.topic)
    // 开始执行分区的删除操作
    topics.foreach { topic =>
        this.onPartitionDeletion(partitionReplicaAssignmentByTopic(topic).keySet)
    }
}

private def startReplicaDeletion(replicasForTopicsToBeDeleted: Set[PartitionAndReplica]) {
    replicasForTopicsToBeDeleted.groupBy(_.topic).keys.foreach { topic =>
        // 获取 topic 名下所有可用的副本集合
        val aliveReplicasForTopic = controllerContext.allLiveReplicas().filter(p => p.topic == topic)
        // 获取 topic 名下中所有的不可用副本集合
        val deadReplicasForTopic = replicasForTopicsToBeDeleted -- aliveReplicasForTopic
        // 获取 topic 名下所有已完成删除的副本集合
        val successfullyDeletedReplicas = controller.replicaStateMachine.replicasInState(topic, ReplicaDeletionSuccessful)
        // 获取 topic 名下未完成删除的 topic 集合
        val replicasForDeletionRetry = aliveReplicasForTopic -- successfullyDeletedReplicas
        // 将不可用副本状态变更为 ReplicaDeletionIneligible
        replicaStateMachine.handleStateChanges(deadReplicasForTopic, ReplicaDeletionIneligible)
        // 将待删除的副本状态变更为 OfflineReplica，用于关闭 follower 对于 leader 的 fetch 请求
        replicaStateMachine.handleStateChanges(replicasForDeletionRetry, OfflineReplica)
        debug("Deletion started for replicas %s".format(replicasForDeletionRetry.mkString(",")))
        // 将待删除的副本状态变更为 ReplicaDeletionStarted，标记当前副本准备好开始删除
        controller.replicaStateMachine.handleStateChanges(replicasForDeletionRetry, ReplicaDeletionStarted,
            new Callbacks.CallbackBuilder().stopReplicaCallback(deleteTopicStopReplicaCallback).build)
        // 如果 topic 存在不可用的副本，标记该 topic 不可删除
        if (deadReplicasForTopic.nonEmpty) {
            debug("Dead Replicas (%s) found for topic %s".format(deadReplicasForTopic.mkString(","), topic))
            markTopicIneligibleForDeletion(Set(topic))
        }
    }
}

private def deleteTopicStopReplicaCallback(stopReplicaResponseObj: AbstractResponse, replicaId: Int) {
    val stopReplicaResponse = stopReplicaResponseObj.asInstanceOf[StopReplicaResponse]
    debug("Delete topic callback invoked for %s".format(stopReplicaResponse))
    val responseMap = stopReplicaResponse.responses.asScala
    // 获取删除失败的分区集合
    val partitionsInError =
        if (stopReplicaResponse.errorCode != Errors.NONE.code) responseMap.keySet
        else responseMap.filter { case (_, error) => error != Errors.NONE.code }.keySet
    // 获取删除失败的副本集合
    val replicasInError = partitionsInError.map(p => PartitionAndReplica(p.topic, p.partition, replicaId))
    inLock(controllerContext.controllerLock) {
        // 将删除失败的副本状态切换成 ReplicaDeletionIneligible，并唤醒删除线程再次尝试删除
        this.failReplicaDeletion(replicasInError)
        // 存在某些副本被成功删除，将这些副本状态切换成 ReplicaDeletionSuccessful
        if (replicasInError.size != responseMap.size) {
            val deletedReplicas = responseMap.keySet -- partitionsInError // 已经成功删除的副本
            this.completeReplicaDeletion(deletedReplicas.map(p => PartitionAndReplica(p.topic, p.partition, replicaId)))
        }
    }
}

def initiateReassignReplicasForTopicPartition(topicAndPartition: TopicAndPartition, reassignedPartitionContext: ReassignedPartitionsContext) {
    // 获取新分配的 AR 集合：RAR
    val newReplicas = reassignedPartitionContext.newReplicas
    val topic = topicAndPartition.topic
    val partition = topicAndPartition.partition
    try {
        // 获取指定 topic 分区之前的 AR 集合：OAR
        val assignedReplicasOpt = controllerContext.partitionReplicaAssignment.get(topicAndPartition)
        assignedReplicasOpt match {
            case Some(assignedReplicas) =>
                // 如果新旧副本未发生变化（OAR == RAR），则无需再分配
                if (assignedReplicas == newReplicas) {
                    throw new KafkaException("Partition %s to be reassigned is already assigned to replicas".format(topicAndPartition) + " %s. Ignoring request for partition reassignment".format(newReplicas.mkString(",")))
                } else {
                    info("Handling reassignment of partition %s to new replicas %s".format(topicAndPartition, newReplicas.mkString(",")))
                    // 为分区注册一个 ReassignedPartitionsIsrChangeListener 监听器
                    this.watchIsrChangesForReassignedPartition(topic, partition, reassignedPartitionContext)
                    controllerContext.partitionsBeingReassigned.put(topicAndPartition, reassignedPartitionContext)
                    // 标记 topic 为不可删除，因为需要执行副本再分配操作
                    deleteTopicManager.markTopicIneligibleForDeletion(Set(topic))
                    // 执行副本再分配操作
                    this.onPartitionReassignment(topicAndPartition, reassignedPartitionContext)
                }
            case None => throw new KafkaException("Attempt to reassign partition %s that doesn't exist".format(topicAndPartition))
        }
    } catch {
        // ... 省略异常处理
    }
}

def onPartitionReassignment(topicAndPartition: TopicAndPartition, reassignedPartitionContext: ReassignedPartitionsContext) {
    // 获取新分配的副本集合 RAR
    val reassignedReplicas = reassignedPartitionContext.newReplicas
    // 如果新分配的副本集合存在一个或多个副本不位于 ISR 集合中
    if (!this.areReplicasInIsr(topicAndPartition.topic, topicAndPartition.partition, reassignedReplicas)) {
        info("New replicas %s for partition %s being ".format(reassignedReplicas.mkString(","), topicAndPartition) + "reassigned not yet caught up with the leader")
        // 获取新添加的副本集合：RAR - OAR
        val newReplicasNotInOldReplicaList = reassignedReplicas.toSet -- controllerContext.partitionReplicaAssignment(topicAndPartition).toSet
        // 获取全部的副本集合：RAR + OAR
        val newAndOldReplicas = (reassignedPartitionContext.newReplicas ++ controllerContext.partitionReplicaAssignment(topicAndPartition)).toSet
        // 1. 将 topic 的全部副本集合（OAR + RAR）更新到 Controller 上下文和 ZK
        this.updateAssignedReplicasForPartition(topicAndPartition, newAndOldReplicas.toSeq)
        // 2. 向 topic 的全部副本集合（OAR + RAR）所在的 broker 节点发送 LeaderAndIsrRequest 请求
        this.updateLeaderEpochAndSendRequest(topicAndPartition, controllerContext.partitionReplicaAssignment(topicAndPartition), newAndOldReplicas.toSeq)
        // 3. 将新增的副本（RAR - OAR）状态切换成 NewReplica
        this.startNewReplicasForReassignedPartition(topicAndPartition, reassignedPartitionContext, newReplicasNotInOldReplicaList)
        info("Waiting for new replicas %s for partition %s being ".format(reassignedReplicas.mkString(","), topicAndPartition) + "reassigned to catch up with the leader")
    }
    // 如果新分配的副本全部位于 ISR 集合中
    else {
        // 1. 获取旧的副本集合（OAR - RAR）
        val oldReplicas = controllerContext.partitionReplicaAssignment(topicAndPartition).toSet -- reassignedReplicas.toSet
        // 2. 将新分配的副本集合（RAR）中所有副本的状态切换成 OnlineReplica
        reassignedReplicas.foreach { replica =>
            replicaStateMachine.handleStateChanges(Set(PartitionAndReplica(topicAndPartition.topic, topicAndPartition.partition, replica)), OnlineReplica)
        }
        // 3. 更新本地记录的分区 AR 集合（OAR -> RAR），同时按需选择新的 leader 副本，并通知到集群中相关节点
        this.moveReassignedPartitionLeaderIfRequired(topicAndPartition, reassignedPartitionContext)
        // 4. 将旧的副本状态切换成 NonExistentReplica
        this.stopOldReplicasOfReassignedPartition(topicAndPartition, reassignedPartitionContext, oldReplicas)
        // 5. 更新 ZK 中记录对应分区的 AR 集合（RAR）
        this.updateAssignedReplicasForPartition(topicAndPartition, reassignedReplicas)
        // 6. 将对应 topic 分区的副本再分配信息从 ZK 和上下文中移除
        this.removePartitionFromReassignedPartitions(topicAndPartition)
        info("Removed partition %s from the list of reassigned partitions in zookeeper".format(topicAndPartition))
        controllerContext.partitionsBeingReassigned.remove(topicAndPartition)
        // 7. 向所有可用的 broker 节点发送 UpdateMetadataRequest 请求，通知副本再分配后的集群状态信息
        this.sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq, Set(topicAndPartition))
        // 8. 取消相关 topic 的不可删除标记，并唤醒 DeleteTopicsThread 线程
        deleteTopicManager.resumeDeletionForTopics(Set(topicAndPartition.topic))
    }
}

def handleNewSession() {
    info("ZK expired; shut down all controller components and try to re-elect")
    // 如果 ZK 上记录的 controller leader 不是当前 broker 节点
    if (controllerElector.getControllerID != config.brokerId) {
        // 尝试清理 controller 之前的状态
        onControllerResignation()
        inLock(controllerContext.controllerLock) {
            // 尝试竞选成为新的 controller leader
            controllerElector.elect
        }
    } else {
        info("ZK expired, but the current controller id %d is the same as this broker id, skip re-elect".format(config.brokerId))
    }
}

def onControllerResignation() {
    debug("Controller resigning, broker id %d".format(config.brokerId))

    // 取消 ZK 上的监听器
    this.deregisterIsrChangeNotificationListener()
    this.deregisterReassignedPartitionsListener()
    this.deregisterPreferredReplicaElectionListener()

    // 关闭 topic 删除机制
    if (deleteTopicManager != null) deleteTopicManager.shutdown()

    // 关闭 partition-rebalance 分区再平衡定时任务
    if (config.autoLeaderRebalanceEnable) autoRebalanceScheduler.shutdown()

    inLock(controllerContext.controllerLock) {
        // 取消所有的 ReassignedPartitionsIsrChangeListener 监听器
        this.deregisterReassignedPartitionsIsrChangeListeners()
        // 关闭分区状态机
        partitionStateMachine.shutdown()
        // 关闭副本状态机
        replicaStateMachine.shutdown()
        // 关闭 ControllerChannelManager，断开与集群中其他 broker 节点之间的连接
        if (controllerContext.controllerChannelManager != null) {
            controllerContext.controllerChannelManager.shutdown()
            controllerContext.controllerChannelManager = null
        }
        // 清除 controller 年代信息
        controllerContext.epoch = 0
        controllerContext.epochZkVersion = 0
        // 切换 broker 节点的状态
        brokerState.newState(RunningAsBroker)

        info("Broker %d resigned as the controller".format(config.brokerId))
    }
}

def doHandleChildChange(parentPath: String, children: Seq[String]) {
    inLock(controllerContext.controllerLock) {
        if (hasStarted.get) {
            try {
                // 获取 /brokers/topics 路径下的子节点，即当前有效的 topic 集合
                val currentChildren = {
                    debug("Topic change listener fired for path %s with children %s".format(parentPath, children.mkString(",")))
                    children.toSet
                }
                // 获取新添加的 topic 集合
                val newTopics = currentChildren -- controllerContext.allTopics
                // 获取已删除的 topic 集合
                val deletedTopics = controllerContext.allTopics -- currentChildren
                // 更新本地记录的所有 topic 集合
                controllerContext.allTopics = currentChildren

                // 从 ZK 读取新增分区的 AR 集合，路径：/brokers/topics/{topic_name}
                val addedPartitionReplicaAssignment = zkUtils.getReplicaAssignmentForTopics(newTopics.toSeq)
                // 更新上下文中记录的每个分区对应的 AR 集合
                controllerContext.partitionReplicaAssignment =
                        controllerContext.partitionReplicaAssignment.filter(p => !deletedTopics.contains(p._1.topic))
                controllerContext.partitionReplicaAssignment ++= addedPartitionReplicaAssignment
                info("New topics: [%s], deleted topics: [%s], new partition replica assignment [%s]".format(newTopics, deletedTopics, addedPartitionReplicaAssignment))
                // 处理新增的 topic，及其新增的分区
                if (newTopics.nonEmpty)
                    controller.onNewTopicCreation(newTopics, addedPartitionReplicaAssignment.keySet)
            } catch {
                case e: Throwable => error("Error while handling new topic", e)
            }
        }
    }
}

def onNewTopicCreation(topics: Set[String], newPartitions: Set[TopicAndPartition]) {
    info("New topic creation callback for %s".format(newPartitions.mkString(",")))
    // 为每个新增的 topic 注册一个 PartitionModificationsListener 监听器
    topics.foreach(topic => partitionStateMachine.registerPartitionChangeListener(topic))
    // 切换新增分区及其副本状态
    this.onNewPartitionCreation(newPartitions)
}

def onNewPartitionCreation(newPartitions: Set[TopicAndPartition]) {
    info("New partition creation callback for %s".format(newPartitions.mkString(",")))
    // 将所有新增的分区状态转换为 NewPartition
    partitionStateMachine.handleStateChanges(newPartitions, NewPartition)
    // 将新增分区的所有副本都转换为 NewReplica，向副本所在 broker 节点发送 LeaderAndIsrRequest 请求，并发送 UpdateMetadataRequest 给所有可用的 broker 节点
    replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions), NewReplica)
    // 为所有新增的分区分配 leader 副本和 ISR 集合，并将分区状态转换为 OnlinePartition
    partitionStateMachine.handleStateChanges(newPartitions, OnlinePartition, offlinePartitionSelector)
    // 将新增分区的所有副本状态转换为 OnlineReplica，并尝试将副本添加到 AR 集合中
    replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions), OnlineReplica)
}

def doHandleChildChange(parentPath: String, children: Seq[String]) {
    inLock(controllerContext.controllerLock) {
        // 从 ZK 获取待删除的 topic 集合
        var topicsToBeDeleted = children.toSet
        debug("Delete topics listener fired for topics %s to be deleted".format(topicsToBeDeleted.mkString(",")))
        // 检查 topic 是否存在，对于不存在的 topic 直接将其从 /admin/delete_topics 路径下删除
        val nonExistentTopics = topicsToBeDeleted -- controllerContext.allTopics
        if (nonExistentTopics.nonEmpty) {
            warn("Ignoring request to delete non-existing topics " + nonExistentTopics.mkString(","))
            nonExistentTopics.foreach(topic => zkUtils.deletePathRecursive(getDeleteTopicPath(topic)))
        }
        topicsToBeDeleted --= nonExistentTopics

        // 如果允许删除 topic，对应 delete.topic.enable 配置
        if (controller.config.deleteTopicEnable) {
            if (topicsToBeDeleted.nonEmpty) {
                info("Starting topic deletion for topics " + topicsToBeDeleted.mkString(","))
                // 检查待删除的 topic 是否处于不可删除的情况
                topicsToBeDeleted.foreach { topic =>
                    // 1. 检测待删除的 topic 是否有分区正在进行优先副本选举
                    val preferredReplicaElectionInProgress =
                        controllerContext.partitionsUndergoingPreferredReplicaElection.map(_.topic).contains(topic)
                    // 2. 检测待删除的 topic 是否有分区正在进行副本再分配
                    val partitionReassignmentInProgress =
                        controllerContext.partitionsBeingReassigned.keySet.map(_.topic).contains(topic)
                    // 如果满足上述 2 个条件之一，则将 topic 标记为不可删除
                    if (preferredReplicaElectionInProgress || partitionReassignmentInProgress)
                        controller.deleteTopicManager.markTopicIneligibleForDeletion(Set(topic))
                }
                // 将可删除的 topic 提交给 TopicDeletionManager 执行删除操作
                controller.deleteTopicManager.enqueueTopicsForDeletion(topicsToBeDeleted)
            }
        } else {
            // 如果配置不允许删除 topic，则从 ZK 上删除对应的节点（/admin/delete_topics）
            for (topic <- topicsToBeDeleted) {
                info("Removing " + getDeleteTopicPath(topic) + " since delete topic is disabled")
                zkUtils.zkClient.delete(getDeleteTopicPath(topic))
            }
        }
    }
}

def doHandleChildChange(parentPath: String, currentBrokerList: Seq[String]) {
    info("Broker change listener fired for path %s with children %s".format(parentPath, currentBrokerList.sorted.mkString(",")))
    inLock(controllerContext.controllerLock) {
        if (hasStarted.get) {
            ControllerStats.leaderElectionTimer.time {
                try {
                    // 从 ZK 获取 broker 节点列表
                    val curBrokers = currentBrokerList.map(_.toInt).toSet.flatMap(zkUtils.getBrokerInfo)
                    val curBrokerIds = curBrokers.map(_.id)
                    val liveOrShuttingDownBrokerIds = controllerContext.liveOrShuttingDownBrokerIds
                    // 筛选新增的 broker 节点列表
                    val newBrokerIds = curBrokerIds -- liveOrShuttingDownBrokerIds
                    // 筛选故障的 broker 节点列表
                    val deadBrokerIds = liveOrShuttingDownBrokerIds -- curBrokerIds
                    val newBrokers = curBrokers.filter(broker => newBrokerIds(broker.id))
                    // 更新 Controller 上下文信息
                    controllerContext.liveBrokers = curBrokers
                    val newBrokerIdsSorted = newBrokerIds.toSeq.sorted
                    val deadBrokerIdsSorted = deadBrokerIds.toSeq.sorted
                    val liveBrokerIdsSorted = curBrokerIds.toSeq.sorted
                    info("Newly added brokers: %s, deleted brokers: %s, all live brokers: %s"
                            .format(newBrokerIdsSorted.mkString(","), deadBrokerIdsSorted.mkString(","), liveBrokerIdsSorted.mkString(",")))
                    // 创建 controller 到新增的 broker 节点之间的网络连接，并启动请求发送线程
                    newBrokers.foreach(controllerContext.controllerChannelManager.addBroker)
                    // 关闭 controller 到故障的 broker 节点之间的网络连接
                    deadBrokerIds.foreach(controllerContext.controllerChannelManager.removeBroker)
                    // 如果存在新增的 broker 节点，通知集群中的其它 broker 节点，并上线新增的分区副本等
                    if (newBrokerIds.nonEmpty) controller.onBrokerStartup(newBrokerIdsSorted)
                    // 如果存在故障的 broker 节点，则下线故障 broker 节点上的分区和副本，并通知到集群中的其它 broker 节点
                    if (deadBrokerIds.nonEmpty) controller.onBrokerFailure(deadBrokerIdsSorted)
                } catch {
                    case e: Throwable => error("Error while handling broker changes", e)
                }
            }
        }
    }
}

def onBrokerStartup(newBrokers: Seq[Int]) {
    info("New broker startup callback for %s".format(newBrokers.mkString(",")))
    val newBrokersSet = newBrokers.toSet

    // 1. 向集群中所有可用的 broker 节点发送 UpdateMetadataRequest 请求，发送的是所有的分区信息，通知节点有新的 broker 加入
    this.sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq)

    // 2. 将新增的 broker 节点上的副本状态设置为 OnlineReplica，以上线对外提供服务
    val allReplicasOnNewBrokers = controllerContext.replicasOnBrokers(newBrokersSet)
    replicaStateMachine.handleStateChanges(allReplicasOnNewBrokers, OnlineReplica)

    // 3. 尝试将状态为 OfflinePartition 和 NewPartition 的分区设置为 OnlinePartition，以触发失效分区的 leader 副本选举
    partitionStateMachine.triggerOnlinePartitionStateChange()

    // 4. 检查正在重新分配副本的分区是否需要重新分配副本
    val partitionsWithReplicasOnNewBrokers = controllerContext.partitionsBeingReassigned.filter {
        case (_, reassignmentContext) => reassignmentContext.newReplicas.exists(newBrokersSet.contains)
    }
    partitionsWithReplicasOnNewBrokers.foreach(p => onPartitionReassignment(p._1, p._2))

    // 5. 如果新增 broker 上有待删除的 topic 的副本，则唤醒 DeleteTopicsThread 线程进行删除，因为对应 topic 之前可能因为该副本失效而被标记为不可删除
    val replicasForTopicsToBeDeleted = allReplicasOnNewBrokers.filter(p => deleteTopicManager.isTopicQueuedUpForDeletion(p.topic))
    if (replicasForTopicsToBeDeleted.nonEmpty) {
        info("Some replicas %s for topics scheduled for deletion %s are on the newly restarted brokers %s. Signaling restart of topic deletion for these topics"
                .format(replicasForTopicsToBeDeleted.mkString(","), deleteTopicManager.topicsToBeDeleted.mkString(","), newBrokers.mkString(",")))
        deleteTopicManager.resumeDeletionForTopics(replicasForTopicsToBeDeleted.map(_.topic))
    }
}

def onBrokerFailure(deadBrokers: Seq[Int]) {
    info("Broker failure callback for %s".format(deadBrokers.mkString(",")))
    // 移除正在关闭的 broker 节点
    val deadBrokersThatWereShuttingDown = deadBrokers.filter(id => controllerContext.shuttingDownBrokerIds.remove(id))
    info("Removed %s from list of shutting down brokers.".format(deadBrokersThatWereShuttingDown))
    val deadBrokersSet = deadBrokers.toSet
    // 1. 如果分区 leader 副本在故障 broker 节点上，将分区状态设置为 OfflinePartition
    val partitionsWithoutLeader = controllerContext.partitionLeadershipInfo.filter(partitionAndLeader =>
        deadBrokersSet.contains(partitionAndLeader._2.leaderAndIsr.leader) &&
                !deleteTopicManager.isTopicQueuedUpForDeletion(partitionAndLeader._1.topic)).keySet
    partitionStateMachine.handleStateChanges(partitionsWithoutLeader, OfflinePartition)

    // 2. 尝试将 OfflinePartition 状态的分区切换成 OnlinePartition 状态
    partitionStateMachine.triggerOnlinePartitionStateChange()

    // 3. 获取分配给故障 broker 节点的副本集合，将这些副本设置为 OfflineReplica 状态
    val allReplicasOnDeadBrokers = controllerContext.replicasOnBrokers(deadBrokersSet)
    val activeReplicasOnDeadBrokers = allReplicasOnDeadBrokers.filterNot(p => deleteTopicManager.isTopicQueuedUpForDeletion(p.topic))
    replicaStateMachine.handleStateChanges(activeReplicasOnDeadBrokers, OfflineReplica)

    // 4. 检查故障 broker 节点上是否有待删除的 topic 副本，如果存在则将其状态转换成 ReplicaDeletionIneligible 状态，并标记 topic 不可删除
    val replicasForTopicsToBeDeleted = allReplicasOnDeadBrokers.filter(p => deleteTopicManager.isTopicQueuedUpForDeletion(p.topic))
    if (replicasForTopicsToBeDeleted.nonEmpty) {
        deleteTopicManager.failReplicaDeletion(replicasForTopicsToBeDeleted)
    }

    // 5. 向所有可用 broker 节点发送 UpdateMetadataRequest 请求，通知部分分区已经失效
    if (partitionsWithoutLeader.isEmpty) {
        this.sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq)
    }
}

def doHandleChildChange(parentPath: String, currentChildren: Seq[String]): Unit = {
    inLock(controller.controllerContext.controllerLock) {
        debug("ISR change notification listener fired")
        try {
            // 从 ZK 上读取 ISR 集合发生变更的 topic 分区集合
            val topicAndPartitions = currentChildren.flatMap(getTopicAndPartition).toSet
            if (topicAndPartitions.nonEmpty) {
                // 从 ZK 读取指定分区的 leader 副本、ISR 集合等信息，更新 Controller 上下文
                controller.updateLeaderAndIsrCache(topicAndPartitions)
                // 向集群所有可用的 broker 节点发送 UpdateMetadataRequest 请求，更新对应 broker 节点缓存的集群元数据信息
                processUpdateNotifications(topicAndPartitions)
            }
        } finally {
            // 删除 /isr_change_notification/partitions 路径下已经处理的信息
            currentChildren.map(x => controller.controllerContext.zkUtils.deletePath(ZkUtils.IsrChangeNotificationPath + "/" + x))
        }
    }
}

private def processUpdateNotifications(topicAndPartitions: immutable.Set[TopicAndPartition]) {
    val liveBrokers: Seq[Int] = controller.controllerContext.liveOrShuttingDownBrokerIds.toSeq
    debug("Sending MetadataRequest to Brokers:" + liveBrokers + " for TopicAndPartitions:" + topicAndPartitions)
    controller.sendUpdateMetadataRequest(liveBrokers, topicAndPartitions)
}

def handleDataChange(dataPath: String, data: Object) {
    val shouldResign = inLock(controllerContext.controllerLock) {
        val amILeaderBeforeDataChange = amILeader
        // 更新本地记录的新的 controller leader 的 ID
        leaderId = KafkaController.parseControllerId(data.toString)
        info("New leader is %d".format(leaderId))
        // 之前是 leader，但是现在切换成了 follower 角色
        amILeaderBeforeDataChange && !amILeader
    }
    // 如果当前 broker 由 leader 变为 follower，则需要执行相应的清理工作
    if (shouldResign) onResigningAsLeader()
}

def handleDataDeleted(dataPath: String) {
    val shouldResign = inLock(controllerContext.controllerLock) {
        debug("%s leader change listener fired for path %s to handle data deleted: trying to elect as a leader".format(brokerId, dataPath))
        amILeader
    }

    // 如果 ZK 上记录的 leader 节点被删除，且当前节点之前是 leader，则需要执行相应的清理工作
    if (shouldResign) onResigningAsLeader()

    // 尝试竞选成为新的 leader
    inLock(controllerContext.controllerLock) {
        elect
    }
}

def doHandleDataChange(dataPath: String, data: AnyRef) {
    inLock(controllerContext.controllerLock) {
        try {
            info(s"Partition modification triggered $data for path $dataPath")
            // 从 ZK 获取 topic 的分区和副本信息
            val partitionReplicaAssignment = zkUtils.getReplicaAssignmentForTopics(List(topic))
            // 筛选新增的分区和副本信息
            val partitionsToBeAdded = partitionReplicaAssignment
                    .filter(p => !controllerContext.partitionReplicaAssignment.contains(p._1))
            // 如果 topic 待删除
            if (controller.deleteTopicManager.isTopicQueuedUpForDeletion(topic))
                error("Skipping adding partitions %s for topic %s since it is currently being deleted".format(partitionsToBeAdded.map(_._1.partition).mkString(","), topic))
            else {
                // 对于正常运行的 topic 的新增分区，更新分区的 AR 集合
                if (partitionsToBeAdded.nonEmpty) {
                    info("New partitions to be added %s".format(partitionsToBeAdded))
                    // 将新增的分区信息添加到 controller 上下文中
                    controllerContext.partitionReplicaAssignment ++= partitionsToBeAdded
                    // 切换新增分区及其副本的状态，使其上线对外提供服务
                    controller.onNewPartitionCreation(partitionsToBeAdded.keySet)
                }
            }
        } catch {
            case e: Throwable => error("Error while handling add partitions for data path " + dataPath, e)
        }
    }
}

def doHandleDataChange(dataPath: String, data: AnyRef) {
    inLock(controllerContext.controllerLock) {
        // 获取需要进行优先副本选举的 topic 分区集合
        val partitionsForPreferredReplicaElection = PreferredReplicaLeaderElectionCommand.parsePreferredReplicaElectionData(data.toString)
        // 过滤已经处于优先副本选举的分区
        val partitions = partitionsForPreferredReplicaElection -- controllerContext.partitionsUndergoingPreferredReplicaElection
        // 过滤掉待删除的 topic 分区
        val partitionsForTopicsToBeDeleted = partitions.filter(p => controller.deleteTopicManager.isTopicQueuedUpForDeletion(p.topic))
        // 对剩余的分区执行优先副本选举
        controller.onPreferredReplicaElection(partitions -- partitionsForTopicsToBeDeleted)
    }
}

def onPreferredReplicaElection(partitions: Set[TopicAndPartition], isTriggeredByAutoRebalance: Boolean = false) {
    info("Starting preferred replica leader election for partitions %s".format(partitions.mkString(",")))
    try {
        // 将分区添加到参与优先副本选举的分区集合中
        controllerContext.partitionsUndergoingPreferredReplicaElection ++= partitions
        // 设置对应的 topic 为不可删除
        deleteTopicManager.markTopicIneligibleForDeletion(partitions.map(_.topic))
        // 设置分区为 OnlinePartition 状态，并使用优先副本选择器重选 leader 副本，同时更新 ZK 和发送 LeaderAndIsrRequest 和 UpdateMetadataRequest 请求
        partitionStateMachine.handleStateChanges(partitions, OnlinePartition, preferredReplicaPartitionLeaderSelector)
    } catch {
        case e: Throwable => error("Error completing preferred replica leader election for partitions %s".format(partitions.mkString(",")), e)
    } finally {
        // 清理 ZK 和上下文中记录的相关数据
        this.removePartitionsFromPreferredReplicaElection(partitions, isTriggeredByAutoRebalance)
        // 将 topic 恢复为可删除，并唤醒 DeleteTopicsThread 线程
        deleteTopicManager.resumeDeletionForTopics(partitions.map(_.topic))
    }
}

def doHandleDataChange(dataPath: String, data: AnyRef) {
    debug("Partitions reassigned listener fired for path %s. Record partitions to be reassigned %s".format(dataPath, data))
    // 从 ZK 读取分区副本的再分配信息
    val partitionsReassignmentData = ZkUtils.parsePartitionReassignmentData(data.toString)
    // 过滤掉正在进行再分配的分区集合
    val partitionsToBeReassigned = inLock(controllerContext.controllerLock) {
        partitionsReassignmentData.filterNot(p => controllerContext.partitionsBeingReassigned.contains(p._1))
    }
    partitionsToBeReassigned.foreach { partitionToBeReassigned =>
        inLock(controllerContext.controllerLock) {
            // 检测 topic 是否为待删除的 topic，如果是的话则放弃对名下分区的再分配操作
            if (controller.deleteTopicManager.isTopicQueuedUpForDeletion(partitionToBeReassigned._1.topic)) {
                error("Skipping reassignment of partition %s for topic %s since it is currently being deleted".format(partitionToBeReassigned._1, partitionToBeReassigned._1.topic))
                controller.removePartitionFromReassignedPartitions(partitionToBeReassigned._1)
            } else {
                val context = ReassignedPartitionsContext(partitionToBeReassigned._2)
                // 为副本再分配做一些前期准备工作
                controller.initiateReassignReplicasForTopicPartition(partitionToBeReassigned._1, context)
            }
        }
    }
}

def doHandleDataChange(dataPath: String, data: AnyRef) {
    inLock(controllerContext.controllerLock) {
        debug("Reassigned partitions isr change listener fired for path %s with children %s".format(dataPath, data))
        val topicAndPartition = TopicAndPartition(topic, partition)
        try {
            controllerContext.partitionsBeingReassigned.get(topicAndPartition) match {
                // 对应的 topic 分区正在执行副本再分配操作
                case Some(reassignedPartitionContext) =>
                    // 从 ZK 上获取对应 topic 分区的 leader 副本和 ISR 集合
                    val newLeaderAndIsrOpt = zkUtils.getLeaderAndIsrForPartition(topic, partition)
                    newLeaderAndIsrOpt match {
                        case Some(leaderAndIsr) => // check if new replicas have joined ISR
                            val caughtUpReplicas = reassignedReplicas & leaderAndIsr.isr.toSet
                            // 如果 RAR 中的副本已经全部进入 ISR 集合中
                            if (caughtUpReplicas == reassignedReplicas) {
                                // 执行副本再分配的后续操作
                                controller.onPartitionReassignment(topicAndPartition, reassignedPartitionContext)
                            } else {
                                // 啥也不干，等待下一次回调
                            }
                        case None => error("Error handling reassignment of partition %s to replicas %s as it was never created".format(topicAndPartition, reassignedReplicas.mkString(",")))
                    }
                // 对应的 topic 分区已经完成副本再分配操作
                case None =>
            }
        } catch {
            case e: Throwable => error("Error while handling partition reassignment", e)
        }
    }
}

class ZookeeperLeaderElector(controllerContext: ControllerContext, // Controller 上下文对象
                             electionPath: String, // /controller
                             onBecomingLeader: () => Unit, // KafkaController#onControllerFailover
                             onResigningAsLeader: () => Unit, // KafkaController#onControllerResignation
                             brokerId: Int, // broker 节点 ID
                             time: Time // 时间戳工具类
                            ) extends LeaderElector with Logging {

    /** 当前 controller leader 的 ID */
    var leaderId: Int = -1
    /** 监听 ZK 的 /controller 节点的数据变化 */
    val leaderChangeListener = new LeaderChangeListener

    // ... 省略方法定义

}

def startup {
    inLock(controllerContext.controllerLock) {
        // 注册 ZK 监听器，监听 /controller 节点下的数据变更
        controllerContext.zkUtils.zkClient.subscribeDataChanges(electionPath, leaderChangeListener)
        // 执行 leader 选举
        elect
    }
}

def elect: Boolean = {
    val timestamp = time.milliseconds.toString
    val electString = Json.encode(Map("version" -> 1, "brokerid" -> brokerId, "timestamp" -> timestamp))

    // 获取 ZK 中记录的 controller leader 的 ID
    leaderId = this.getControllerID
    // 已经存在 controller leader，放弃选举
    if (leaderId != -1) {
        debug("Broker %d has been elected as leader, so stopping the election process.".format(leaderId))
        return amILeader
    }

    try {
        // 尝试创建 ZK 临时节点，如果临时节点已经存在，则抛出异常
        val zkCheckedEphemeral = new ZKCheckedEphemeral(electionPath,
            electString,
            controllerContext.zkUtils.zkConnection.getZookeeper,
            JaasUtils.isZkSecurityEnabled)
        zkCheckedEphemeral.create()
        info(brokerId + " successfully elected as leader")
        // 创建成功，更新 leader 节点 ID
        leaderId = brokerId
        // 回调
        onBecomingLeader()
    } catch {
        // leader 已经存在
        case _: ZkNodeExistsException =>
            // If someone else has written the path, then
            leaderId = getControllerID
        case e2: Throwable =>
            error("Error while electing or becoming leader on broker %d".format(brokerId), e2)
            // 重置 leaderId，并删除 /controller 节点
            resign()
    }
    // 检测当前 broker 节点是否成为 leader
    amILeader
}

def onControllerFailover() {
    if (isRunning) {
        info("Broker %d starting become controller state transition".format(config.brokerId))
        // 1. 从 ZK 读取历史 controller 年代信息，并更新 Controller 上下文
        this.readControllerEpochFromZookeeper()
        // 2. 递增 controller 年代信息，并更新到 ZK
        this.incrementControllerEpoch(zkUtils.zkClient)

        // 3. 注册 ZK 监听器

        this.registerReassignedPartitionsListener() // 注册 PartitionsReassignedListener
        this.registerIsrChangeNotificationListener() // 注册 IsrChangeNotificationListener
        this.registerPreferredReplicaElectionListener() // 注册 PreferredReplicaElectionListener
        partitionStateMachine.registerListeners() // 注册 TopicChangeListener 和 DeleteTopicsListener
        replicaStateMachine.registerListeners() // 注册 BrokerChangeListener

        // 4. 初始化 Controller 上下文信息
        this.initializeControllerContext()

        // 5. 向集群中所有可用的 broker 发送 UpdateMetadataRequest 请求，更新对应节点缓存的集群元数据
        this.sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq)

        // 6. 启动副本状态机，并初始化各个副本的状态
        replicaStateMachine.startup()

        // 7. 启动分区状态机，并初始化各个分区的状态
        partitionStateMachine.startup()

        // 8. 为所有的 topic 注册 PartitionModificationsListener 监听器
        controllerContext.allTopics.foreach(topic => partitionStateMachine.registerPartitionChangeListener(topic))
        info("Broker %d is ready to serve as the new controller with epoch %d".format(config.brokerId, epoch))

        // 9. 处理副本需要再分配的分区
        this.maybeTriggerPartitionReassignment()

        // 10. 处理需要进行优先副本选举的分区
        this.maybeTriggerPreferredReplicaElection()

        // 11. 依据配置决定是否开启分区自动均衡的功能
        if (config.autoLeaderRebalanceEnable) {
            info("starting the partition rebalance scheduler")
            // 启用定时任务，周期性检测
            autoRebalanceScheduler.startup()
            autoRebalanceScheduler.schedule(
                "partition-rebalance-thread",
                checkAndTriggerPartitionRebalance,
                5,
                config.leaderImbalanceCheckIntervalSeconds.toLong,
                TimeUnit.SECONDS)
        }
        // 12. 启动 TopicDeletionManager，用于对指定的 topic 执行删除操作
        deleteTopicManager.start()
    } else
          info("Controller has been shut down, aborting startup/failover")
}

private def initializeControllerContext() {
    // 读取 /brokers/ids 节点，初始化可用的 broker 集合
    controllerContext.liveBrokers = zkUtils.getAllBrokersInCluster.toSet
    // 读取 /brokers/topics 节点，初始化集群中全部的 topic 集合
    controllerContext.allTopics = zkUtils.getAllTopics.toSet
    // 读取 /brokers/topics/{topic_name}/partitions 节点，初始化每个分区的 AR 集合
    controllerContext.partitionReplicaAssignment = zkUtils.getReplicaAssignmentForTopics(controllerContext.allTopics.toSeq)
    controllerContext.partitionLeadershipInfo = new mutable.HashMap[TopicAndPartition, LeaderIsrAndControllerEpoch]
    controllerContext.shuttingDownBrokerIds = mutable.Set.empty[Int]
    // 读取 /brokers/topics/{topic_name}/partition/{partitionId}/stat 节点，初始化每个 topic 分区的 leader 副本和 ISR 集合等信息
    this.updateLeaderAndIsrCache()
    // 启动 ControllerChannelManager，用于建立到集群中所有 broker 节点的连接，并与之通信
    this.startChannelManager()
    // 读取 /admin/preferred_replica_election 节点，初始化需要执行优先副本选举的分区
    this.initializePreferredReplicaElection()
    // 读取 /admin/reassign_partitions 节点，初始化需要进行副本重新分配的分区
    this.initializePartitionReassignment()
    // 启动 TopicDeletionManager，用于管理待删除的 topic 和不可删除的 topic 集合
    this.initializeTopicDeletion()
}

private def checkAndTriggerPartitionRebalance(): Unit = {
    if (isActive) {
        trace("checking need to trigger partition rebalance")
        // 获取所有可用的副本集合，key 是优先副本所在的 broker 节点 ID
        var preferredReplicasForTopicsByBrokers: Map[Int, Map[TopicAndPartition, Seq[Int]]] = null
        inLock(controllerContext.controllerLock) {
            // 获取优先副本所在的 broker 节点 ID 与分区的对应关系
            preferredReplicasForTopicsByBrokers =
                    controllerContext.partitionReplicaAssignment
                            // 过滤掉待删除的 topic
                            .filterNot(p => deleteTopicManager.isTopicQueuedUpForDeletion(p._1.topic))
                            // 按照第一个副本（优先副本） ID 进行分组
                            .groupBy { case (_, assignedReplicas) => assignedReplicas.head }
        }
        debug("preferred replicas by broker " + preferredReplicasForTopicsByBrokers)

        // 计算每个 broker 节点的副本不均衡比率（imbalanceRatio）
        preferredReplicasForTopicsByBrokers.foreach { case (leaderBroker, topicAndPartitionsForBroker) =>
            var imbalanceRatio: Double = 0
            // 计算存在 leader 副本，但不是以优先副本作为 leader 副本的分区副本集合
            var topicsNotInPreferredReplica: Map[TopicAndPartition, Seq[Int]] = null
            inLock(controllerContext.controllerLock) {
                topicsNotInPreferredReplica = topicAndPartitionsForBroker.filter { case (topicPartition, _) =>
                    controllerContext.partitionLeadershipInfo.contains(topicPartition) && // 存在 leader 副本
                            controllerContext.partitionLeadershipInfo(topicPartition).leaderAndIsr.leader != leaderBroker // 且 leader 副本不是优先副本
                }
                debug("topics not in preferred replica " + topicsNotInPreferredReplica)
                val totalTopicPartitionsForBroker = topicAndPartitionsForBroker.size
                val totalTopicPartitionsNotLedByBroker = topicsNotInPreferredReplica.size
                // 计算当前 broker 节点的 imbalance 比率：（不是以优先副本作为 leader 副本的分区数 / 所在 broker 节点的总分区数）
                imbalanceRatio = totalTopicPartitionsNotLedByBroker.toDouble / totalTopicPartitionsForBroker
                trace("leader imbalance ratio for broker %d is %f".format(leaderBroker, imbalanceRatio))
            }

            // 当 broker 节点上的不均衡比率大于阈值（对应 leader.imbalance.per.broker.percentage 配置）时，触发优先副本选举机制
            if (imbalanceRatio > (config.leaderImbalancePerBrokerPercentage.toDouble / 100)) {
                topicsNotInPreferredReplica.keys.foreach { topicPartition =>
                    inLock(controllerContext.controllerLock) {
                        // 对满足条件的 topic 分区，执行优先副本选举
                        if (controllerContext.liveBrokerIds.contains(leaderBroker) && // 对应的 broker 节点是有效的
                                controllerContext.partitionsBeingReassigned.isEmpty && // 没有分区正在执行副本再分配
                                controllerContext.partitionsUndergoingPreferredReplicaElection.isEmpty && // 没有分区正在执行优先副本选举
                                !deleteTopicManager.isTopicQueuedUpForDeletion(topicPartition.topic) && // 分区所属 topic 正常运行
                                controllerContext.allTopics.contains(topicPartition.topic)) { // 分区所属 topic 是有效的
                            // 执行优先副本选举
                            onPreferredReplicaElection(Set(topicPartition), isTriggeredByAutoRebalance = true)
                        }
                    }
                }
            }
        }
    }
}

def shutdownBroker(id: Int): Set[TopicAndPartition] = {
    // 保证当前 Controller 是 leader 角色
    if (!isActive) {
        throw new ControllerMovedException("Controller moved to another broker. Aborting controlled shutdown")
    }

    controllerContext.brokerShutdownLock synchronized {
        info("Shutting down broker " + id)
        inLock(controllerContext.controllerLock) {
            // 校验目标 broker 节点是否处于运行中，对于不存在或已经关闭的 broker 节点不需要执行关闭操作
            if (!controllerContext.liveOrShuttingDownBrokerIds.contains(id)) throw new BrokerNotAvailableException("Broker id %d does not exist.".format(id))
            // 记录正在关闭的 broker 节点 ID
            controllerContext.shuttingDownBrokerIds.add(id)
        }

        // 获取待关闭 broker 节点上所有的分区和副本信息
        val allPartitionsAndReplicationFactorOnBroker: Set[(TopicAndPartition, Int)] =
            inLock(controllerContext.controllerLock) {
                controllerContext.partitionsOnBroker(id).map(topicAndPartition =>
                    (topicAndPartition, controllerContext.partitionReplicaAssignment(topicAndPartition).size))
            }

        // 遍历处理待关闭 broker 节点上的分区和副本
        allPartitionsAndReplicationFactorOnBroker.foreach {
            case (topicAndPartition, replicationFactor) =>
                // Move leadership serially to relinquish lock.
                inLock(controllerContext.controllerLock) {
                    controllerContext.partitionLeadershipInfo.get(topicAndPartition).foreach { currLeaderIsrAndControllerEpoch =>
                        // 如果开启了副本机制
                        if (replicationFactor > 1) {
                            // 如果分区 leader 副本位于待关闭的 broker 节点上
                            if (currLeaderIsrAndControllerEpoch.leaderAndIsr.leader == id) {
                                // 使用 ControlledShutdownLeaderSelector 选择器重新为分区选择新的 leader 和 ISR 集合，
                                // 并将结果写入 ZK，然后发送 LeaderAndIsrRequest 和 UpdateMetadataRequest 请求给集群中相应 broker 节点
                                partitionStateMachine.handleStateChanges(
                                    Set(topicAndPartition), OnlinePartition, controlledShutdownPartitionLeaderSelector)
                            }
                            // 如果分区 leader 副本不位于待关闭的 broker 节点上
                            else {
                                try {
                                    // 发送 StopReplicaRequest 请求给待关闭的 broker 节点，关闭分区位于该节点上的副本（不删除副本）
                                    brokerRequestBatch.newBatch()
                                    brokerRequestBatch.addStopReplicaRequestForBrokers(
                                        Seq(id), topicAndPartition.topic, topicAndPartition.partition, deletePartition = false)
                                    brokerRequestBatch.sendRequestsToBrokers(epoch)
                                } catch {
                                    // ... 省略异常处理
                                }
                                // 将副本状态切换成 OfflineReplica，并尝试从 ISR 集合中移除，同时通知到集群中相应 broker 节点
                                replicaStateMachine.handleStateChanges(
                                    Set(PartitionAndReplica(topicAndPartition.topic, topicAndPartition.partition, id)), OfflineReplica)
                            }
                        }
                    }
                }
        }

        // 统计 leader 副本依然处于待关闭 broker 节点上的分区数目
        def replicatedPartitionsBrokerLeads(): Iterable[TopicAndPartition] = inLock(controllerContext.controllerLock) {
            trace("All leaders = " + controllerContext.partitionLeadershipInfo.mkString(","))
            controllerContext.partitionLeadershipInfo.filter {
                case (topicAndPartition, leaderIsrAndControllerEpoch) =>
                    leaderIsrAndControllerEpoch.leaderAndIsr.leader == id &&
                            controllerContext.partitionReplicaAssignment(topicAndPartition).size > 1
            }.keys
        }

        replicatedPartitionsBrokerLeads().toSet
    }
}