Kafka 源码解析：分区多副本容错机制

class Replica(val brokerId: Int, // 当前副本所在的 broker 的 ID
              val partition: Partition, // 当前副本所属的 topic 分区对象
              time: Time = Time.SYSTEM, // 时间戳工具
              initialHighWatermarkValue: Long = 0L, // 初始 HW 值
              val log: Option[Log] = None // 当前副本所属的 Log 对象，如果是远程副本，该字段为空，通过该字段可以区分是本地副本还是远程副本
             ) extends Logging {

    /**
     * 记录副本的 HW 值，消费者只能读取 HW 之前的消息，之后的消息对消费者不可见，
     * 由 leader 副本维护，当消息被 ISR 集合中所有副本成功同步时更新该字段。
     */
    @volatile private[this] var highWatermarkMetadata = new LogOffsetMetadata(initialHighWatermarkValue)
    /**
     * 记录副本所属 Log 对象最后一条消息的 offset 值：
     * - 如果是本地副本，可以直接从 Log#nextOffsetMetadata 字段中获取；
     * - 如果是远程副本，则由其它 broker 发送请求来更新该值。
     */
    @volatile private[this] var logEndOffsetMetadata = LogOffsetMetadata.UnknownOffsetMetadata
    /** 缓存上次从 leader 拉取消息时 leader 副本的 LEO 值 */
    @volatile private[this] var lastFetchLeaderLogEndOffset = 0L
    /** 记录上次从 leader 拉取消息的时间戳 */
    @volatile private[this] var lastFetchTimeMs = 0L
    /** 记录当前 follower 从 leader 拉取消息的最近一次时间戳，用于标识当前 follower 滞后 leader 的程度 */
    @volatile private[this] var _lastCaughtUpTimeMs = 0L
    /** 副本所属 topic 分区对象 */
    val topicPartition: TopicPartition = partition.topicPartition

    // ... 省略方法定义

}

def updateLogReadResult(logReadResult: LogReadResult) {
    // 更新 _lastCaughtUpTimeMs 值，记录了 follower 从 leader 拉取消息的最新时间
    if (logReadResult.info.fetchOffsetMetadata.messageOffset >= logReadResult.leaderLogEndOffset)
        _lastCaughtUpTimeMs = math.max(_lastCaughtUpTimeMs, logReadResult.fetchTimeMs)
    else if (logReadResult.info.fetchOffsetMetadata.messageOffset >= lastFetchLeaderLogEndOffset)
             _lastCaughtUpTimeMs = math.max(_lastCaughtUpTimeMs, lastFetchTimeMs)

    // 如果当前副本是远程副本，则更新当前副本的 LEO 值
    logEndOffset = logReadResult.info.fetchOffsetMetadata
    // 更新本地记录的从 leader 拉取消息时 leader 副本的 LEO 值
    lastFetchLeaderLogEndOffset = logReadResult.leaderLogEndOffset
    // 更新本地记录的从 leader 拉取消息的时间戳
    lastFetchTimeMs = logReadResult.fetchTimeMs
}

private def logEndOffset_=(newLogEndOffset: LogOffsetMetadata) {
    if (isLocal) {
        // 如果是本地副本无需更新 LEO 值，而是由对应 Log 对象的 Log#logEndOffsetMetadata 字段决定
        throw new KafkaException(s"Should not set log end offset on partition $topicPartition's local replica $brokerId")
    } else {
        // 如果当前副本是远程副本，则更新副本的 LEO 值
        logEndOffsetMetadata = newLogEndOffset
        trace(s"Setting log end offset for replica $brokerId for partition $topicPartition to [$logEndOffsetMetadata]")
    }
}

def logEndOffset: LogOffsetMetadata = if (isLocal) log.get.logEndOffsetMetadata else logEndOffsetMetadata

def highWatermark_=(newHighWatermark: LogOffsetMetadata) {
    if (isLocal) {
        // 如果是本地副本，则更新对应的 HW 值
        highWatermarkMetadata = newHighWatermark
        trace(s"Setting high watermark for replica $brokerId partition $topicPartition to [$newHighWatermark]")
    } else {
        throw new KafkaException(s"Should not set high watermark on partition $topicPartition's non-local replica $brokerId")
    }
}

def highWatermark: LogOffsetMetadata = highWatermarkMetadata

class Partition(val topic: String, // 分区所属的 topic
                val partitionId: Int, // 分区编号
                time: Time, // 时间戳工具
                replicaManager: ReplicaManager // 副本管理
               ) extends Logging with KafkaMetricsGroup {

    /** topic 分区对象 */
    val topicPartition = new TopicPartition(topic, partitionId)
    /** 当前 broker 的 ID */
    private val localBrokerId = replicaManager.config.brokerId
    /** 管理分区日志数据 */
    private val logManager = replicaManager.logManager
    /** ZK 工具类 */
    private val zkUtils = replicaManager.zkUtils
    /** AR 集合，维护当前分区全部副本的集合，key 是副本 ID */
    private val assignedReplicaMap = new Pool[Int, Replica]
    /** leader 副本的年代信息 */
    @volatile private var leaderEpoch: Int = LeaderAndIsr.initialLeaderEpoch - 1
    /** leader 副本的 ID */
    @volatile var leaderReplicaIdOpt: Option[Int] = None
    /** 当前分区的 ISR 集合 */
    @volatile var inSyncReplicas: Set[Replica] = Set.empty[Replica]
    /** 当前集群控制器的年代信息，会在切换副本角色时进行更新 */
    private var controllerEpoch: Int = KafkaController.InitialControllerEpoch - 1

    // ... 省略方法定义

}

def getOrCreateReplica(replicaId: Int = localBrokerId): Replica = {
    // 尝试从 AR 集合中获取 replicaId 对应的 Replica 对象，如果不存在则创建一个
    assignedReplicaMap.getAndMaybePut(replicaId, {
        // 如果是本地副本
        if (this.isReplicaLocal(replicaId)) {
            // 获取 log 相关配置信息，ZK 中的配置会覆盖默认配置
            val config = LogConfig.fromProps(logManager.defaultConfig.originals, AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic, topic))

            // 创建 topic 分区对应的 Log 对象，如果已经存在则直接返回
            val log = logManager.createLog(topicPartition, config)

            // 加载对应 log 目录下的 replication-offset-checkpoint 文件，其中记录了每个 topic 分区的 HW 值
            val checkpoint = replicaManager.highWatermarkCheckpoints(log.dir.getParentFile.getAbsolutePath)
            val offsetMap = checkpoint.read()

            // 获取当前 topic 分区对应的 HW 值，并与 LEO 比较，选择较小的值作为此副本的 HW 位置
            val offset = math.min(offsetMap.getOrElse(topicPartition, 0L), log.logEndOffset)

            // 创建 Replica 对象
            new Replica(replicaId, this, time, offset, Some(log))
        }
        // 如果是远程副本，无需加载本地对应的日志数据
        else new Replica(replicaId, this, time)
    })
}

def createLog(topicPartition: TopicPartition, config: LogConfig): Log = {
    logCreationOrDeletionLock synchronized {
        // 获取指定 topic 分区对应的 Log 对象
        getLog(topicPartition).getOrElse {
            // 如果存在多个 log 目录，则选择 Log 数目最少的目录
            val dataDir = this.nextLogDir()
            // 创建当前 topic 分区对应的日志目录
            val dir = new File(dataDir, topicPartition.topic + "-" + topicPartition.partition)
            dir.mkdirs()
            // 创建 Log 对象
            val log = new Log(dir, config, recoveryPoint = 0L, scheduler, time)
            // 缓存到本地 logs 字段中
            logs.put(topicPartition, log)
            info("Created log for partition [%s,%d] in %s with properties {%s}."
                    .format(topicPartition.topic, topicPartition.partition, dataDir.getAbsolutePath, config.originals.asScala.mkString(", ")))
            log
        }
    }
}

def makeLeader(controllerId: Int, partitionStateInfo: PartitionState, correlationId: Int): Boolean = {
    val (leaderHWIncremented, isNewLeader) = inWriteLock(leaderIsrUpdateLock) {
        // 1. 更新本地记录的 controller 的年代信息
        controllerEpoch = partitionStateInfo.controllerEpoch

        // 2. 获取/创建请求信息中 AR 和 ISR 集合中所有副本对应的 Replica 对象
        val allReplicas = partitionStateInfo.replicas.asScala.map(_.toInt)
        allReplicas.foreach(replica => getOrCreateReplica(replica))
        val newInSyncReplicas = partitionStateInfo.isr.asScala.map(r => getOrCreateReplica(r)).toSet

        // 3. 移除本地缓存的所有已过期的的副本对象
        (assignedReplicas.map(_.brokerId) -- allReplicas).foreach(removeReplica)

        // 4. 更新本地记录的分区 leader 副本相关信息
        inSyncReplicas = newInSyncReplicas // 更新 ISR 集合
        leaderEpoch = partitionStateInfo.leaderEpoch // 更新 leader 副本的年代信息
        zkVersion = partitionStateInfo.zkVersion // 更新 ZK 的版本信息

        // 5. 检测分区 leader 副本是否发生变化
        val isNewLeader =
            if (leaderReplicaIdOpt.isDefined && leaderReplicaIdOpt.get == localBrokerId) {
                false // 未发生变化
            } else {
                // leader 发生变化，更新分区 leader 副本 ID
                leaderReplicaIdOpt = Some(localBrokerId)
                true
            }

        // 6. 遍历所有的 follower 副本，更新对应副本的相关时间戳信息
        val leaderReplica = getReplica().get // 获取 leader 副本 Replica 对象
        val curLeaderLogEndOffset = leaderReplica.logEndOffset.messageOffset // 获取 leader 副本的 LEO 值
        val curTimeMs = time.milliseconds
        (assignedReplicas - leaderReplica).foreach { replica =>
            val lastCaughtUpTimeMs = if (inSyncReplicas.contains(replica)) curTimeMs else 0L
            replica.resetLastCaughtUpTime(curLeaderLogEndOffset, curTimeMs, lastCaughtUpTimeMs)
        }

        // 7. 如果当前 leader 是新选举出来的，则修正 leader 副本的 HW 值，并重置本地缓存的所有远程副本的相关信息
        if (isNewLeader) {
            // 尝试修正新 leader 副本的 HW 值
            leaderReplica.convertHWToLocalOffsetMetadata()
            // 重置本地缓存的所有远程副本的相关信息
            assignedReplicas.filter(_.brokerId != localBrokerId).foreach(_.updateLogReadResult(LogReadResult.UnknownLogReadResult))
        }

        // 8. 尝试后移 leader 副本的 HW 值
        (maybeIncrementLeaderHW(leaderReplica), isNewLeader)
    }

    // 9. 如果 leader 副本的 HW 值增加了，则尝试执行监听当前 topic 分区的 DelayedFetch 和 DelayedProduce 任务
    if (leaderHWIncremented) tryCompleteDelayedRequests()

    isNewLeader
}

def makeFollower(controllerId: Int, partitionStateInfo: PartitionState, correlationId: Int): Boolean = {
    inWriteLock(leaderIsrUpdateLock) {
        val allReplicas = partitionStateInfo.replicas.asScala.map(_.toInt)
        val newLeaderBrokerId: Int = partitionStateInfo.leader

        // 1. 更新本地记录的 controller 的年代信息
        controllerEpoch = partitionStateInfo.controllerEpoch

        // 2. 获取/创建请求信息中所有副本对应的 Replica 对象
        allReplicas.foreach(r => getOrCreateReplica(r))

        // 3. 移除本地缓存的所有已过期的的副本对象
        (assignedReplicas.map(_.brokerId) -- allReplicas).foreach(removeReplica)

        // 4. 更新本地记录的分区 leader 副本相关信息，其中 ISR 集合由 leader 副本维护，将 follower 副本上的 ISR 集合置空
        inSyncReplicas = Set.empty[Replica]
        leaderEpoch = partitionStateInfo.leaderEpoch // 更新 leader 副本的年代信息
        zkVersion = partitionStateInfo.zkVersion // 更新 zk 版本信息

        // 5. 检测分区 leader 副本是否发生变化，如果发生变化则更新本地记录的 ID 值
        if (leaderReplicaIdOpt.isDefined && leaderReplicaIdOpt.get == newLeaderBrokerId) {
            false
        } else {
            // 发生变化，更新本地记录的分区 leader 副本的 ID
            leaderReplicaIdOpt = Some(newLeaderBrokerId)
            true
        }
    }
}

def appendRecordsToLeader(records: MemoryRecords, requiredAcks: Int = 0): LogAppendInfo = {
    val (info, leaderHWIncremented) = inReadLock(leaderIsrUpdateLock) {
        leaderReplicaIfLocal match {
            // 只有 leader 副本支持追加消息操作
            case Some(leaderReplica) =>
                // 获取 leader 副本对应的 Log 对象
                val log = leaderReplica.log.get
                // 对应 min.insync.replicas 配置，表示 ISR 集合的最小值
                val minIsr = log.config.minInSyncReplicas
                // 获取当前分区 ISR 集合的大小
                val inSyncSize = inSyncReplicas.size

                // 如果用户指定 acks = -1，但是当前 ISR 集合小于允许的最小值，则不允许追加消息，防止数据丢失
                if (inSyncSize < minIsr && requiredAcks == -1) {
                    throw new NotEnoughReplicasException(
                        "Number of insync replicas for partition %s is [%d], below required minimum [%s]".format(topicPartition, inSyncSize, minIsr))
                }

                // 往 leader 副本的 Log 对象中追加消息
                val info = log.append(records)
                // 有新的日志数据被追加，尝试执行监听当前 topic 分区的 DelayedFetch 延时任务
                replicaManager.tryCompleteDelayedFetch(TopicPartitionOperationKey(this.topic, this.partitionId))
                // 尝试后移 leader 副本的 HW 值
                (info, maybeIncrementLeaderHW(leaderReplica))

            // 如果不是 leader 副本，则抛出异常
            case None =>
                throw new NotLeaderForPartitionException("Leader not local for partition %s on broker %d".format(topicPartition, localBrokerId))
        }
    }

    // 如果 leader 副本的 HW 值增加了，则尝试执行监听当前 topic 分区的 DelayedFetch 和 DelayedProduce 任务
    if (leaderHWIncremented) tryCompleteDelayedRequests()

    info
}

def maybeExpandIsr(replicaId: Int, logReadResult: LogReadResult) {
    val leaderHWIncremented = inWriteLock(leaderIsrUpdateLock) {
        leaderReplicaIfLocal match {
            // 只有当本地副本是 leader 副本时，才执行扩张操作，因为 ISR 集合由 leader 副本维护
            case Some(leaderReplica) =>
                // 获取目标 follower 副本对应的 Replica 对象
                val replica = getReplica(replicaId).get
                // 获取 leader 副本对应的 HW 值
                val leaderHW = leaderReplica.highWatermark
                // 判断当前 follower 是否应该被加入到 ISR 集合，并在成功加入后更新相关信息
                if (!inSyncReplicas.contains(replica) // follower 副本不在 ISR 集合中
                        && assignedReplicas.map(_.brokerId).contains(replicaId) // AR 集合中包含该 follower 副本
                        && replica.logEndOffset.offsetDiff(leaderHW) >= 0) { // follower 副本的 LEO 已经追赶上 leader 副本的 HW 值
                    // 将 follower 副本添加到 ISR 集合中
                    val newInSyncReplicas = inSyncReplicas + replica
                    info(s"Expanding ISR for partition $topicPartition from ${inSyncReplicas.map(_.brokerId).mkString(",")} to ${newInSyncReplicas.map(_.brokerId).mkString(",")}")
                    // 更新 ZK 和本地记录的新的 ISR 集合信息
                    this.updateIsr(newInSyncReplicas)
                    replicaManager.isrExpandRate.mark()
                }
                // 尝试后移 leader 副本的 HW 值
                this.maybeIncrementLeaderHW(leaderReplica, logReadResult.fetchTimeMs)

            // 如果不是 leader 副本，啥也不干
            case None => false
        }
    }

    // 如果 leader 副本的 HW 值发生变化，尝试执行监听当前 topic 分区的 DelayedFetch 和 DelayedProduce 延时任务
    if (leaderHWIncremented) tryCompleteDelayedRequests()
}

def maybeShrinkIsr(replicaMaxLagTimeMs: Long) {
    val leaderHWIncremented = inWriteLock(leaderIsrUpdateLock) {
        leaderReplicaIfLocal match {
            // 只有当本地副本是 leader 副本时，才执行缩减操作，因为 ISR 集合由 leader 副本维护
            case Some(leaderReplica) =>
                // 从 ISR 集合中获取滞后的 follower 副本集合
                val outOfSyncReplicas = this.getOutOfSyncReplicas(leaderReplica, replicaMaxLagTimeMs)
                if (outOfSyncReplicas.nonEmpty) {
                    // 将滞后的 follower 副本从 ISR 集合中剔除
                    val newInSyncReplicas = inSyncReplicas -- outOfSyncReplicas
                    assert(newInSyncReplicas.nonEmpty)
                    info("Shrinking ISR for partition [%s,%d] from %s to %s".format(topic, partitionId,
                        inSyncReplicas.map(_.brokerId).mkString(","), newInSyncReplicas.map(_.brokerId).mkString(",")))
                    // 将新的 ISR 集合信息上报给 ZK，同时更新本地记录的 ISR 集合信息
                    this.updateIsr(newInSyncReplicas)
                    replicaManager.isrShrinkRate.mark()
                    // 尝试后移 leader 副本的 HW 值
                    this.maybeIncrementLeaderHW(leaderReplica)
                } else {
                    false
                }
            // 如果不是 leader 副本，则啥也不做
            case None => false
        }
    }

    // 如果 leader 副本的 HW 值发生变化，尝试执行监听当前 topic 分区的 DelayedFetch 和 DelayedProduce 延时任务
    if (leaderHWIncremented) tryCompleteDelayedRequests()
}

def getOutOfSyncReplicas(leaderReplica: Replica, maxLagMs: Long): Set[Replica] = {
    // 获取 ISR 集合中所有的 follower 副本
    val candidateReplicas = inSyncReplicas - leaderReplica
    // 获取超过给定时间（对应 replica.lag.time.max.ms 配置）未向 leader 副本请求拉取消息的 follower 副本集合
    val laggingReplicas = candidateReplicas.filter(r => (time.milliseconds - r.lastCaughtUpTimeMs) > maxLagMs)
    laggingReplicas
}

def checkEnoughReplicasReachOffset(requiredOffset: Long): (Boolean, Errors) = {
    leaderReplicaIfLocal match {
        // 如果当前副本是 leader 副本
        case Some(leaderReplica) =>
            // 获取 ISR 集合
            val curInSyncReplicas = inSyncReplicas

            // 对应 min.insync.replicas 配置
            val minIsr = leaderReplica.log.get.config.minInSyncReplicas

            // 如果当前请求的 offset 小于等于 HW 的 offset
            if (leaderReplica.highWatermark.messageOffset >= requiredOffset) {
                // 如果当前分区的 ISR 集合大小大于等于允许的最小值
                if (minIsr <= curInSyncReplicas.size) (true, Errors.NONE)
                // 否则返回 NOT_ENOUGH_REPLICAS_AFTER_APPEND 错误
                else (true, Errors.NOT_ENOUGH_REPLICAS_AFTER_APPEND)
            } else {
                // 如果当前请求的 offset 大于 HW，则直接返回 false，因为 HW 之后的消息对于客户端不可见
                (false, Errors.NONE)
            }

        // 如果当前副本是 follower 副本，则返回 NOT_LEADER_FOR_PARTITION 错误
        case None => (false, Errors.NOT_LEADER_FOR_PARTITION)
    }
}

private def maybeIncrementLeaderHW(leaderReplica: Replica, curTime: Long = time.milliseconds): Boolean = {
    // 获取位于 ISR 集合中，或最近一次从 leader 拉取消息的时间戳位于指定时间范围（对应 replica.lag.time.max.ms 配置）内的所有副本的 LEO 值
    val allLogEndOffsets = assignedReplicas.filter { replica =>
        curTime - replica.lastCaughtUpTimeMs <= replicaManager.config.replicaLagTimeMaxMs || inSyncReplicas.contains(replica)
    }.map(_.logEndOffset)

    // 以这些副本中最小的 LEO 值作为 leader 副本新的 HW 值
    val newHighWatermark = allLogEndOffsets.min(new LogOffsetMetadata.OffsetOrdering)

    // 比较新旧 HW 值，如果旧的 HW 小于新的 HW，或者旧的 HW 对应的 LogSegment 的 baseOffset 小于新的 HW 的 LogSegment 对象的 baseOffset，则更新
    val oldHighWatermark = leaderReplica.highWatermark
    if (oldHighWatermark.messageOffset < newHighWatermark.messageOffset || oldHighWatermark.onOlderSegment(newHighWatermark)) {
        leaderReplica.highWatermark = newHighWatermark
        debug("High watermark for partition [%s,%d] updated to %s".format(topic, partitionId, newHighWatermark))
        true
    } else {
        debug("Skipping update high watermark since Old hw %s is larger than new hw %s for partition [%s,%d]. All leo's are %s"
                .format(oldHighWatermark, newHighWatermark, topic, partitionId, allLogEndOffsets.mkString(",")))
        false
    }
}

class ReplicaManager(val config: KafkaConfig, // 相关配置对象
                     metrics: Metrics,
                     time: Time, // 时间戳工具
                     val zkUtils: ZkUtils, // ZK 工具类
                     scheduler: Scheduler, // 定时任务调度器
                     val logManager: LogManager, // 用于对分区日志数据执行读写操作
                     val isShuttingDown: AtomicBoolean, // 标记 kafka 服务是否正在执行关闭操作
                     quotaManager: ReplicationQuotaManager,
                     threadNamePrefix: Option[String] = None) extends Logging with KafkaMetricsGroup {

    /**
     * 记录 kafka controller 的年代信息，当重新选择 controller leader 时会递增该字段，
     * 用于校验来自 controller 的请求的年代信息，防止处理来自老的 controller 的请求
     */
    @volatile var controllerEpoch: Int = KafkaController.InitialControllerEpoch - 1
    /** 本地 broker 的 ID */
    private val localBrokerId = config.brokerId
    /** 记录当前 broker 管理的所有分区信息，如果不存在则创建 */
    private val allPartitions = new Pool[TopicPartition, Partition](Some(tp => new Partition(tp.topic, tp.partition, time, this)))
    /** 管理向 leader 副本发送 FetchRequest 请求的 ReplicaFetcherThread 线程 */
    val replicaFetcherManager = new ReplicaFetcherManager(config, this, metrics, time, threadNamePrefix, quotaManager)
    /** 标记 highwatermark-checkpoint 定时任务是否已经启动 */
    private val highWatermarkCheckPointThreadStarted = new AtomicBoolean(false)
    /** 记录每个 log 目录与对应 topic 分区 HW 值的映射关系 */
    val highWatermarkCheckpoints: Predef.Map[String, OffsetCheckpoint] = config.logDirs.map(dir =>
        (new File(dir).getAbsolutePath, new OffsetCheckpoint(new File(dir, ReplicaManager.HighWatermarkFilename)))).toMap
    /** 标记 highwatermark-checkpoint 定时任务是否已经启动 */
    private var hwThreadInitialized = false
    /** 记录 ISR 集合发生变化的 topic 分区信息 */
    private val isrChangeSet: mutable.Set[TopicPartition] = new mutable.HashSet[TopicPartition]()
    private val lastIsrChangeMs = new AtomicLong(System.currentTimeMillis())
    private val lastIsrPropagationMs = new AtomicLong(System.currentTimeMillis())
    /** 管理 DelayedProduce 延时任务的炼狱 */
    val delayedProducePurgatory: DelayedOperationPurgatory[DelayedProduce] = DelayedOperationPurgatory[DelayedProduce](
        purgatoryName = "Produce", localBrokerId, config.producerPurgatoryPurgeIntervalRequests)
    /** 管理 DelayedFetch 延时任务的炼狱 */
    val delayedFetchPurgatory: DelayedOperationPurgatory[DelayedFetch] = DelayedOperationPurgatory[DelayedFetch](
        purgatoryName = "Fetch", localBrokerId, config.fetchPurgatoryPurgeIntervalRequests)

        // ... 省略相关方法定义

}

def startup() {
    // 定时检测当前 broker 节点管理的每个分区是否需要缩减 ISR 集合，并执行缩减操作
    scheduler.schedule("isr-expiration", maybeShrinkIsr, period = config.replicaLagTimeMaxMs / 2, unit = TimeUnit.MILLISECONDS)
    // 定时将 ISR 集合发生变化的 topic 分区记录到 ZK
    scheduler.schedule("isr-change-propagation", maybePropagateIsrChanges, period = 2500L, unit = TimeUnit.MILLISECONDS)
}

def maybePropagateIsrChanges() {
    val now = System.currentTimeMillis()
    isrChangeSet synchronized {
        // 定期将 ISR 集合发生变化的分区记录到 ZK，kafka controller 对相应 ZK 路径添加了 Watcher，
        // 当 Watcher 被触发后会向所有可用的 broker 节点发送 UpdateMetadataRequest 请求，以更新 broker 节点缓存的所有分区状态信息
        if (isrChangeSet.nonEmpty &&
                // 最后一次有 ISR 集合发生变化的时间距离现在已经超过 5 秒
                (lastIsrChangeMs.get() + ReplicaManager.IsrChangePropagationBlackOut < now
                        // 上次写入 ZK 的时间距离现在已经超过 1 分钟
                        || lastIsrPropagationMs.get() + ReplicaManager.IsrChangePropagationInterval < now)) {
            // 将 ISR 集合发生变更的 topic 分区信息记录到 ZK
            ReplicationUtils.propagateIsrChanges(zkUtils, isrChangeSet)
            isrChangeSet.clear()
            lastIsrPropagationMs.set(now)
        }
    }
}

def maybeUpdateMetadataCache(correlationId: Int,
                             updateMetadataRequest: UpdateMetadataRequest,
                             metadataCache: MetadataCache): Seq[TopicPartition] = {
    replicaStateChangeLock synchronized {
        // 校验 controller 的年代信息，避免处理来自已经过期的 controller 的请求
        if (updateMetadataRequest.controllerEpoch < controllerEpoch) {
            val stateControllerEpochErrorMessage = ("Broker %d received update metadata request with correlation id %d from an " +
                    "old controller %d with epoch %d. Latest known controller epoch is %d").format(localBrokerId,
                correlationId, updateMetadataRequest.controllerId, updateMetadataRequest.controllerEpoch, controllerEpoch)
            stateChangeLogger.warn(stateControllerEpochErrorMessage)
            throw new ControllerMovedException(stateControllerEpochErrorMessage)
        } else {
            // 更新所有分区的状态信息，并返回需要被移除的分区集合
            val deletedPartitions = metadataCache.updateCache(correlationId, updateMetadataRequest)
            // 更新本地缓存的 controller 年代信息
            controllerEpoch = updateMetadataRequest.controllerEpoch
            deletedPartitions
        }
    }
}

def startHighWaterMarksCheckPointThread(): Unit = {
    if (highWatermarkCheckPointThreadStarted.compareAndSet(false, true))
        scheduler.schedule(
            "highwatermark-checkpoint",
            checkpointHighWatermarks,
            period = config.replicaHighWatermarkCheckpointIntervalMs,
            unit = TimeUnit.MILLISECONDS)
}

def checkpointHighWatermarks() {
    // 获取所有分区全部的本地副本 Replica 对象
    val replicas = allPartitions.values.flatMap(_.getReplica(localBrokerId))
    // 按照副本所在的 log 目录进行分组
    val replicasByDir = replicas.filter(_.log.isDefined).groupBy(_.log.get.dir.getParentFile.getAbsolutePath)
    // 遍历将位于相同 log 目录下的分区 HW 值，写入到对应的 replication-offset-checkpoint 文件中
    for ((dir, reps) <- replicasByDir) {
        // 获取每个 topic 分区对应的 HW 值
        val hwms: Map[TopicPartition, Long] = reps.map(r => r.partition.topicPartition -> r.highWatermark.messageOffset).toMap
        try {
            // 更新对应 log 目录下的 replication-offset-checkpoint 文件
            highWatermarkCheckpoints(dir).write(hwms)
        } catch {
            case e: IOException =>
                fatal("Error writing to highwatermark file: ", e)
                Runtime.getRuntime.halt(1)
        }
    }
}

def addFetcherForPartitions(partitionAndOffsets: Map[TopicPartition, BrokerAndInitialOffset]) {
    mapLock synchronized {
        val partitionsPerFetcher = partitionAndOffsets.groupBy {
            case (topicPartition, brokerAndInitialOffset) =>
                // 由分区所属的 topic 和分区编号计算得到对应的 fetcher 线程 ID，并与 broker 的网络位置信息组成 key，然后按 key 进行分组，
                // 后面会为每组分配一个 fetcher 线程，每个线程只连接一个 broker，可以同时为组内多个分区的 follower 副本执行同步操作。
                BrokerAndFetcherId(brokerAndInitialOffset.broker, this.getFetcherId(topicPartition.topic, topicPartition.partition))
        }

        // 启动所有的的 fetcher 线程，如果对应线程不存在，则创建并启动
        for ((brokerAndFetcherId, partitionAndOffsets) <- partitionsPerFetcher) {
            var fetcherThread: AbstractFetcherThread = null
            fetcherThreadMap.get(brokerAndFetcherId) match {
                case Some(f) => fetcherThread = f
                case None =>
                    // 创建 ReplicaFetcherThread 线程对象，并记录到 fetcherThreadMap 集合中
                    fetcherThread = this.createFetcherThread(brokerAndFetcherId.fetcherId, brokerAndFetcherId.broker)
                    fetcherThreadMap.put(brokerAndFetcherId, fetcherThread)
                    fetcherThread.start() // 启动线程
            }

            // 将 topic 分区和同步起始位置传递给 fetcher 线程，并唤醒 fetcher 线程开始同步
            fetcherThreadMap(brokerAndFetcherId).addPartitions(partitionAndOffsets.map {
                case (tp, brokerAndInitOffset) => tp -> brokerAndInitOffset.initOffset
            })
        }
    }
}

override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): AbstractFetcherThread = {
    val threadName = threadNamePrefix match {
        case None => "ReplicaFetcherThread-%d-%d".format(fetcherId, sourceBroker.id)
        case Some(p) => "%s:ReplicaFetcherThread-%d-%d".format(p, fetcherId, sourceBroker.id)
    }
    new ReplicaFetcherThread(threadName, fetcherId, sourceBroker, brokerConfig, replicaMgr, metrics, time, quotaManager)
}

override def doWork() {
    val fetchRequest = inLock(partitionMapLock) {
        // 创建 FetchRequest 请求对象
        val fetchRequest = this.buildFetchRequest(partitionStates.partitionStates.asScala.map { state =>
            state.topicPartition -> state.value
        })
        // 如果没有拉取消息的需求，则等待一会后重试
        if (fetchRequest.isEmpty) {
            trace("There are no active partitions. Back off for %d ms before sending a fetch request".format(fetchBackOffMs))
            partitionMapCond.await(fetchBackOffMs, TimeUnit.MILLISECONDS)
        }
        fetchRequest
    }
    // 发送 FetchRequest 请求，并处理 FetchResponse 响应
    if (!fetchRequest.isEmpty) this.processFetchRequest(fetchRequest)
}

private def processFetchRequest(fetchRequest: REQ) {
    val partitionsWithError = mutable.Set[TopicPartition]()
    var responseData: Seq[(TopicPartition, PD)] = Seq.empty

    // 1. 发送 FetchRequest 请求，并阻塞等待响应
    try {
        trace("Issuing to broker %d of fetch request %s".format(sourceBroker.id, fetchRequest))
        responseData = this.fetch(fetchRequest) // 模板方法
    } catch {
        // ... 省略异常处理
    }
    fetcherStats.requestRate.mark()

    // 2. 处理响应
    if (responseData.nonEmpty) {
        inLock(partitionMapLock) {
            // 遍历处理每个 topic 分区对应的响应
            responseData.foreach { case (topicPartition, partitionData) =>
                val topic = topicPartition.topic
                val partitionId = topicPartition.partition
                Option(partitionStates.stateValue(topicPartition)).foreach(currentPartitionFetchState =>
                    // 如果从发送 FetchRequest 请求到收到响应期间，offset 没有发生变化，则追加收到的日志数据
                    if (fetchRequest.offset(topicPartition) == currentPartitionFetchState.offset) {
                        Errors.forCode(partitionData.errorCode) match {
                            case Errors.NONE =>
                                try {
                                    // 获取返回的消息集合
                                    val records = partitionData.toRecords
                                    // 获取返回的最后一条消息的 offset 值
                                    val newOffset = records.shallowEntries.asScala.lastOption.map(_.nextOffset).getOrElse(currentPartitionFetchState.offset)

                                    fetcherLagStats.getAndMaybePut(topic, partitionId).lag = Math.max(0L, partitionData.highWatermark - newOffset)
                                    // 将从 leader 副本获取到的消息追加到当前 follower 副本对应的 Log 对象中
                                    this.processPartitionData(topicPartition, currentPartitionFetchState.offset, partitionData)

                                    val validBytes = records.validBytes
                                    if (validBytes > 0) {
                                        // 更新本地缓存的 fetch 状态
                                        partitionStates.updateAndMoveToEnd(topicPartition, new PartitionFetchState(newOffset))
                                        fetcherStats.byteRate.mark(validBytes)
                                    }
                                } catch {
                                    // ... 省略异常处理
                                }
                            // follower 请求的 offset 超出了 leader 的 LEO 值
                            case Errors.OFFSET_OUT_OF_RANGE =>
                                try {
                                    // 计算有效的 offset，并更新本地缓存的 fetch 状态
                                    val newOffset = this.handleOffsetOutOfRange(topicPartition)
                                    partitionStates.updateAndMoveToEnd(topicPartition, new PartitionFetchState(newOffset))
                                    error("Current offset %d for partition [%s,%d] out of range; reset offset to %d".format(currentPartitionFetchState.offset, topic, partitionId, newOffset))
                                } catch {
                                    // ... 省略异常处理
                                }
                            // ... 其他异常
                        }
                    })
            }
        }
    }

    // 对于操作存在异常的 topic 分区，暂停发送 FetchRequest 请求，休息一会儿
    if (partitionsWithError.nonEmpty) {
        debug("handling partitions with error for %s".format(partitionsWithError))
        this.handlePartitionsWithErrors(partitionsWithError)
    }
}

def processPartitionData(topicPartition: TopicPartition, fetchOffset: Long, partitionData: PartitionData) {
    try {
        val replica = replicaMgr.getReplica(topicPartition).get
        val records = partitionData.toRecords

        // 如果拉取到的消息数据过大，则打印异常
        this.maybeWarnIfOversizedRecords(records, topicPartition)

        if (fetchOffset != replica.logEndOffset.messageOffset)
            throw new RuntimeException("Offset mismatch for partition %s: fetched offset = %d, log end offset = %d.".format(topicPartition, fetchOffset, replica.logEndOffset.messageOffset))

        // 将消息追加到 Log 中，因为 leader 已经为消息分配了 offset，所以 follower 无需在对消息分配 offset 值
        replica.log.get.append(records, assignOffsets = false)

        // 更新对应 follower 副本的 HW 值
        val followerHighWatermark = replica.logEndOffset.messageOffset.min(partitionData.highWatermark)
        replica.highWatermark = new LogOffsetMetadata(followerHighWatermark)

        if (quota.isThrottled(topicPartition)) quota.record(records.sizeInBytes)
    } catch {
        case e: KafkaStorageException =>
            fatal(s"Disk error while replicating data for $topicPartition", e)
            Runtime.getRuntime.halt(1)
    }
}

def handleOffsetOutOfRange(topicPartition: TopicPartition): Long = {
    val replica = replicaMgr.getReplica(topicPartition).get

    // 发送 ListOffsetRequest 请求，获取 leader 副本的 LEO 值
    val leaderEndOffset: Long = this.earliestOrLatestOffset(topicPartition, ListOffsetRequest.LATEST_TIMESTAMP, brokerConfig.brokerId)

    // 如果 leader 副本的 LEO 值落后于 follower 副本的 LEO 值
    if (leaderEndOffset < replica.logEndOffset.messageOffset) {
        // 依据配置（unclean.leader.election.enable）决定是否需要停机
        if (!LogConfig.fromProps(brokerConfig.originals,
            AdminUtils.fetchEntityConfig(replicaMgr.zkUtils, ConfigType.Topic, topicPartition.topic)).uncleanLeaderElectionEnable) {
            // Log a fatal error and shutdown the broker to ensure that data loss does not unexpectedly occur.
            fatal("Exiting because log truncation is not allowed for partition %s,".format(topicPartition) +
                    " Current leader %d's latest offset %d is less than replica %d's latest offset %d"
                            .format(sourceBroker.id, leaderEndOffset, brokerConfig.brokerId, replica.logEndOffset.messageOffset))
            System.exit(1)
        }

        warn("Replica %d for partition %s reset its fetch offset from %d to current leader %d's latest offset %d"
                .format(brokerConfig.brokerId, topicPartition, replica.logEndOffset.messageOffset, sourceBroker.id, leaderEndOffset))
        // 将分区对应的 Log 截断到 leader 副本的 LEO 位置，从该位置开始重新与 leader 副本进行同步
        replicaMgr.logManager.truncateTo(Map(topicPartition -> leaderEndOffset))

        // 返回下次获取消息的 offset 位置
        leaderEndOffset
    } else {
        // ... 第 2 种情况
    }
}

def handleOffsetOutOfRange(topicPartition: TopicPartition): Long = {
    val replica = replicaMgr.getReplica(topicPartition).get

    // 发送 ListOffsetRequest 请求，获取 leader 副本的 LEO 值
    val leaderEndOffset: Long = this.earliestOrLatestOffset(topicPartition, ListOffsetRequest.LATEST_TIMESTAMP, brokerConfig.brokerId)

    // 如果 leader 副本的 LEO 值落后于 follower 副本的 LEO 值
    if (leaderEndOffset < replica.logEndOffset.messageOffset) {
        // ... 第 1 种情况
    } else {
        // 发送 ListOffsetRequest 请求，获取 leader 副本的 startOffset 值
        val leaderStartOffset: Long = this.earliestOrLatestOffset(topicPartition, ListOffsetRequest.EARLIEST_TIMESTAMP, brokerConfig.brokerId)
        warn("Replica %d for partition %s reset its fetch offset from %d to current leader %d's start offset %d"
                .format(brokerConfig.brokerId, topicPartition, replica.logEndOffset.messageOffset, sourceBroker.id, leaderStartOffset))
        // 选择下次获取消息的起始 offset 值
        val offsetToFetch = Math.max(leaderStartOffset, replica.logEndOffset.messageOffset)
        // 如果当前 leader 的 startOffset 大于对应副本的 LEO 值，则将该副本的 Log 全部截断，并创建新的 activeSegment 对象
        if (leaderStartOffset > replica.logEndOffset.messageOffset)
            replicaMgr.logManager.truncateFullyAndStartAt(topicPartition, leaderStartOffset)

        // 返回下次获取消息的 offset 位置
        offsetToFetch
    }
}

def addPartitions(partitionAndOffsets: Map[TopicPartition, Long]) {
    partitionMapLock.lockInterruptibly()
    try {
        // 基于指定的 offset 构造每个 topic 分区合法的 PartitionFetchState 对象，忽略已经存在的 topic 分区
        val newPartitionToState = partitionAndOffsets
                .filter { case (tp, _) => !partitionStates.contains(tp) }
                .map { case (tp, offset) =>
                    // 基于指定的 offset 创建对应的 PartitionFetchState 对象，如果 offset 无效，则尝试解析得到合法的 offset 值
                    val fetchState =
                        if (PartitionTopicInfo.isOffsetInvalid(offset)) new PartitionFetchState(this.handleOffsetOutOfRange(tp))
                        else new PartitionFetchState(offset)
                    tp -> fetchState
                }
        // 获取并更新本地缓存的已有的 topic 分区与 PartitionFetchState 对象之间的映射关系
        val existingPartitionToState = partitionStates.partitionStates.asScala.map { state => state.topicPartition -> state.value }.toMap
        partitionStates.set((existingPartitionToState ++ newPartitionToState).asJava)
        // 唤醒当前 fetcher 线程，执行同步操作
        partitionMapCond.signalAll()
    } finally {
        partitionMapLock.unlock()
    }
}

def becomeLeaderOrFollower(correlationId: Int,
                           leaderAndISRRequest: LeaderAndIsrRequest,
                           metadataCache: MetadataCache,
                           onLeadershipChange: (Iterable[Partition], Iterable[Partition]) => Unit): BecomeLeaderOrFollowerResult = {

    replicaStateChangeLock synchronized {
        // 用于记录每个分区角色切换操作的状态码
        val responseMap = new mutable.HashMap[TopicPartition, Short]
        // 校验 controller 的年代信息，避免处理来自已经过期的 controller 的请求
        if (leaderAndISRRequest.controllerEpoch < controllerEpoch) {
            BecomeLeaderOrFollowerResult(responseMap, Errors.STALE_CONTROLLER_EPOCH.code)
        } else {
            val controllerId = leaderAndISRRequest.controllerId
            // 1. 更新本地缓存的 kafka controller 的年代信息
            controllerEpoch = leaderAndISRRequest.controllerEpoch

            // 2. 校验请求的 leader 副本的年代信息，以及是否由当前 broker 节点管理，将满足条件的分区信息记录到 partitionState 集合中
            val partitionState = new mutable.HashMap[Partition, PartitionState]()
            leaderAndISRRequest.partitionStates.asScala.foreach { case (topicPartition, stateInfo) =>
                // 获取/创建指定 topic 分区的 Partition 对象
                val partition = this.getOrCreatePartition(topicPartition)
                // 获取 leader 副本的年代信息
                val partitionLeaderEpoch = partition.getLeaderEpoch
                // 校验 leader 副本的年代信息，需要保证请求中的 leader 副本的年代信息大于本地缓存的 topic 分区 leader 副本的年代信息
                if (partitionLeaderEpoch < stateInfo.leaderEpoch) {
                    // 如果请求的分区副本位于当前 broker 节点上，记录到 partitionState 集合中
                    if (stateInfo.replicas.contains(localBrokerId))
                        partitionState.put(partition, stateInfo)
                    else {
                        // 请求的分区副本不在当前 broker 节点上，响应 UNKNOWN_TOPIC_OR_PARTITION 错误
                        responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
                    }
                } else {
                    // 请求中的 leader 副本的年代信息小于等于本地记录的对应 topic 分区 leader 副本的年代信息，响应 STALE_CONTROLLER_EPOCH 错误
                    responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH.code)
                }
            }

            // 3. 将请求对象中的分区集合分割成 leader 和 follower 两类，并执行角色切换
            val partitionsTobeLeader = partitionState.filter { case (_, stateInfo) => stateInfo.leader == localBrokerId }
            val partitionsToBeFollower = partitionState -- partitionsTobeLeader.keys

            // 3.1 将指定分区的副本切换成 leader 角色
            val partitionsBecomeLeader = if (partitionsTobeLeader.nonEmpty)
                                             this.makeLeaders(controllerId, controllerEpoch, partitionsTobeLeader, correlationId, responseMap)
                                         else
                                             Set.empty[Partition]

            // 3.2 将指定分区的副本切换成 follower 角色
            val partitionsBecomeFollower = if (partitionsToBeFollower.nonEmpty)
                                               this.makeFollowers(controllerId, controllerEpoch, partitionsToBeFollower, correlationId, responseMap, metadataCache)
                                           else
                                               Set.empty[Partition]

            // 4. 如果 highwatermark-checkpoint 定时任务尚未启动，则执行启动
            if (!hwThreadInitialized) {
                this.startHighWaterMarksCheckPointThread()
                hwThreadInitialized = true
            }

            // 5. 关闭空闲的 fetcher 线程
            replicaFetcherManager.shutdownIdleFetcherThreads()

            // 6. 执行回调函数，完成 GroupCoordinator 的迁移操作
            onLeadershipChange(partitionsBecomeLeader, partitionsBecomeFollower)

            // 7. 封装结果对象返回
            BecomeLeaderOrFollowerResult(responseMap, Errors.NONE.code)
        }
    }
}

private def makeLeaders(controllerId: Int,
                        epoch: Int,
                        partitionState: Map[Partition, PartitionState], // 记录需要切换成 leader 角色的分区副本信息
                        correlationId: Int,
                        responseMap: mutable.Map[TopicPartition, Short]): Set[Partition] = {

    // 初始化每个 topic 分区的错误码为 NONE
    for (partition <- partitionState.keys)
        responseMap.put(partition.topicPartition, Errors.NONE.code)

    val partitionsToMakeLeaders: mutable.Set[Partition] = mutable.Set()
    try {
        // 如果对应的副本当前是 follower 角色，需要要先停止这些副本的消息同步工作
        replicaFetcherManager.removeFetcherForPartitions(partitionState.keySet.map(_.topicPartition))

        // 遍历处理 partitionState 集合，将其中记录的分区转换成 leader 角色
        partitionState.foreach {
            case (partition, partitionStateInfo) =>
                // 调用 Partition#makeLeader 方法，将分区的本地副本切换成 leader 角色
                if (partition.makeLeader(controllerId, partitionStateInfo, correlationId))
                    partitionsToMakeLeaders += partition // 记录成功完成 leader 角色切换的副本对应的分区
                else
                    // ... 省略日志打点
        }
    } catch {
        // ... 省略异常处理
    }

    partitionsToMakeLeaders
}

private def makeFollowers(controllerId: Int,
                          epoch: Int,
                          partitionState: Map[Partition, PartitionState],
                          correlationId: Int,
                          responseMap: mutable.Map[TopicPartition, Short],
                          metadataCache: MetadataCache): Set[Partition] = {

    // 初始化每个 topic 分区的错误码为 NONE
    for (partition <- partitionState.keys)
        responseMap.put(partition.topicPartition, Errors.NONE.code)

    val partitionsToMakeFollower: mutable.Set[Partition] = mutable.Set()
    try {
        partitionState.foreach {
            case (partition, partitionStateInfo) =>
                // 检测 leader 副本所在的 broker 是否可用
                val newLeaderBrokerId = partitionStateInfo.leader
                metadataCache.getAliveBrokers.find(_.id == newLeaderBrokerId) match {
                    // 仅对 leader 副本所在 broker 节点可用的副本执行角色切换
                    case Some(_) =>
                        // 调用 Partition#makeFollower 方法，将分区的本地副本切换成 follower 角色
                        if (partition.makeFollower(controllerId, partitionStateInfo, correlationId))
                            partitionsToMakeFollower += partition // 记录成功完成 follower 角色切换的副本对应的分区
                        else
                            // ... 省略日志打点
                    // 对应 leader 副本所在的 broker 节点失效
                    case None =>
                        // 即使 leader 副本所在的 broker 不可用，也要创建本地副本对象，主要是为了在 checkpoint 文件中记录此分区的 HW 值
                        partition.getOrCreateReplica()
                }
        }

        // 停止与旧的 leader 副本同步的 fetcher 线程
        replicaFetcherManager.removeFetcherForPartitions(partitionsToMakeFollower.map(_.topicPartition))

        // 由于 leader 副本发生变化，所以新旧 leader 在 [HW, LEO] 之间的消息可能不一致，
        // 但是 HW 之前的消息是一致的，所以将 Log 截断到 HW 位置，可能会出现 unclean leader election 的场景
        logManager.truncateTo(partitionsToMakeFollower.map { partition =>
            (partition.topicPartition, partition.getOrCreateReplica().highWatermark.messageOffset)
        }.toMap)

        // 尝试完成监听对应分区的 DelayedProduce 和 DelayedFetch 延时任务
        partitionsToMakeFollower.foreach { partition =>
            val topicPartitionOperationKey = new TopicPartitionOperationKey(partition.topicPartition)
            this.tryCompleteDelayedProduce(topicPartitionOperationKey)
            this.tryCompleteDelayedFetch(topicPartitionOperationKey)
        }

        // 检测 ReplicaManager 的运行状态
        if (isShuttingDown.get()) {
            // ... 省略日志打点
        }
        // 重新启用与新 leader 副本同步的 fetcher 线程
        else {
            val partitionsToMakeFollowerWithLeaderAndOffset = partitionsToMakeFollower.map(partition =>
                partition.topicPartition -> BrokerAndInitialOffset(
                    metadataCache.getAliveBrokers.find(_.id == partition.leaderReplicaIdOpt.get).get.getBrokerEndPoint(config.interBrokerListenerName),
                    partition.getReplica().get.logEndOffset.messageOffset)).toMap
            // 为需要同步的分区创建并启动同步线程，从指定的 offset 开始与 leader 副本进行同步
            replicaFetcherManager.addFetcherForPartitions(partitionsToMakeFollowerWithLeaderAndOffset)
        }
    } catch {
        // ... 省略异常处理
    }

    partitionsToMakeFollower
}

def stopReplicas(stopReplicaRequest: StopReplicaRequest): (mutable.Map[TopicPartition, Short], Short) = {
    replicaStateChangeLock synchronized {
        val responseMap = new collection.mutable.HashMap[TopicPartition, Short]
        // 校验 controller 的年代信息，避免处理来自已经过期的 controller 的请求
        if (stopReplicaRequest.controllerEpoch() < controllerEpoch) {
            (responseMap, Errors.STALE_CONTROLLER_EPOCH.code)
        } else {
            val partitions = stopReplicaRequest.partitions.asScala
            // 更新本地记录的 kafka controller 的年代信息
            controllerEpoch = stopReplicaRequest.controllerEpoch
            // 停止对指定分区的数据同步 fetcher 线程
            replicaFetcherManager.removeFetcherForPartitions(partitions)
            for (topicPartition <- partitions) {
                // 关闭指定分区的副本
                val errorCode = this.stopReplica(topicPartition, stopReplicaRequest.deletePartitions())
                responseMap.put(topicPartition, errorCode)
            }
            (responseMap, Errors.NONE.code)
        }
    }
}

def stopReplica(topicPartition: TopicPartition, deletePartition: Boolean): Short = {
    val errorCode = Errors.NONE.code
    getPartition(topicPartition) match {
        case Some(_) =>
            // 如果 deletePartition = true，则删除分区对应的副本及其日志和索引文件
            if (deletePartition) {
                // 从本地移除指定的 topic 分区
                val removedPartition = allPartitions.remove(topicPartition)
                if (removedPartition != null) {
                    // 删除分区的日志和索引文件，并清空本地缓存的相关信息
                    removedPartition.delete() // this will delete the local log
                    val topicHasPartitions = allPartitions.keys.exists(tp => topicPartition.topic == tp.topic)
                    if (!topicHasPartitions) BrokerTopicStats.removeMetrics(topicPartition.topic)
                }
            }
        // 本地未缓存对应的分区（一般发生在对应的 topic 已经被删除，但是期间 broker 宕机了），直接尝试对应的删除日志和索引文件
        case None =>
            if (deletePartition && logManager.getLog(topicPartition).isDefined) logManager.asyncDelete(topicPartition)
    }
    errorCode
}

def appendRecords(timeout: Long,
                  requiredAcks: Short,
                  internalTopicsAllowed: Boolean,
                  entriesPerPartition: Map[TopicPartition, MemoryRecords],
                  responseCallback: Map[TopicPartition, PartitionResponse] => Unit) {

    // 如果 acks 参数合法
    if (this.isValidRequiredAcks(requiredAcks)) {
        val sTime = time.milliseconds
        // 将消息追加到 Log 对象中
        val localProduceResults = this.appendToLocalLog(internalTopicsAllowed, entriesPerPartition, requiredAcks)
        debug("Produce to local log in %d ms".format(time.milliseconds - sTime))

        // 封装数据追加结果
        val produceStatus = localProduceResults.map { case (topicPartition, result) =>
            topicPartition -> ProducePartitionStatus(
                result.info.lastOffset + 1, // 下一次请求日志的 offset 值
                new PartitionResponse(result.error, result.info.firstOffset, result.info.logAppendTime)) // response status
        }

        // 如果需要生成 DelayedProduce 延时任务
        if (this.delayedRequestRequired(requiredAcks, entriesPerPartition, localProduceResults)) {
            // 创建 DelayedProduce 延时任务对象，将回调响应函数封装到延时任务对象中
            val produceMetadata = ProduceMetadata(requiredAcks, produceStatus)
            val delayedProduce = new DelayedProduce(timeout, produceMetadata, this, responseCallback)

            // 创建当前延时任务监听的一系列 key 对象，监听本次追加操作的所有 topic 分区
            val producerRequestKeys = entriesPerPartition.keys.map(new TopicPartitionOperationKey(_)).toSeq

            // 尝试执行延时任务，如果还未到期则将任务交由炼狱管理
            delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys)
        } else {
            // 无需生成 DelayedProduce 延时任务，立即响应
            val produceResponseStatus = produceStatus.mapValues(status => status.responseStatus)
            responseCallback(produceResponseStatus)
        }
    } else {
        // 对应的 acks 参数错误，构造 INVALID_REQUIRED_ACKS 响应
        val responseStatus = entriesPerPartition.map { case (topicPartition, _) =>
            topicPartition -> new PartitionResponse(
                Errors.INVALID_REQUIRED_ACKS, LogAppendInfo.UnknownLogAppendInfo.firstOffset, Record.NO_TIMESTAMP)
        }
        // 回调响应
        responseCallback(responseStatus)
    }
}

private def appendToLocalLog(internalTopicsAllowed: Boolean, // 是否允许往内部 topic 追加消息
                             entriesPerPartition: Map[TopicPartition, MemoryRecords], // 对应分区需要追加的消息数据
                             requiredAcks: Short // acks
                            ): Map[TopicPartition, LogAppendResult] = {

    // 遍历处理每个 topic 分区及其待追加的消息数据
    entriesPerPartition.map { case (topicPartition, records) =>
        // 如果追加的对象是内部 topic，依据参数 internalTopicsAllowed 决定是否追加
        if (Topic.isInternal(topicPartition.topic) && !internalTopicsAllowed) {
            (topicPartition, LogAppendResult(
                LogAppendInfo.UnknownLogAppendInfo,
                Some(new InvalidTopicException(s"Cannot append to internal topic ${topicPartition.topic}"))))
        } else {
            try {
                // 获取 topic 分区对应的 Partition 对象
                val partitionOpt = this.getPartition(topicPartition)
                val info = partitionOpt match {
                    // 往 leader 副本对应的 Log 对象中追加消息数据
                    case Some(partition) => partition.appendRecordsToLeader(records, requiredAcks)
                    // 找不到 topic 分区对应的 Partition 对象
                    case None => throw new UnknownTopicOrPartitionException("Partition %s doesn't exist on %d".format(topicPartition, localBrokerId))
                }

                // 返回每个分区写入的消息结果
                (topicPartition, LogAppendResult(info))
            } catch {
                // ... 省略异常处理
            }
        }
    }
}

def fetchMessages(timeout: Long,
                  replicaId: Int,
                  fetchMinBytes: Int,
                  fetchMaxBytes: Int,
                  hardMaxBytesLimit: Boolean,
                  fetchInfos: Seq[(TopicPartition, PartitionData)],
                  quota: ReplicaQuota = UnboundedQuota,
                  responseCallback: Seq[(TopicPartition, FetchPartitionData)] => Unit) {
    // 标记是否是来自 follower 的 fetch 请求
    val isFromFollower = replicaId >= 0
    // 是否只读 leader 副本的消息，一般 debug 模式下可以读 follower 副本的数据
    val fetchOnlyFromLeader: Boolean = replicaId != Request.DebuggingConsumerId
    // 是否只读已完成提交的消息（即 HW 之前的消息），如果是来自消费者的请求则该参数是 true，如果是 follower 则该参数是 false
    val fetchOnlyCommitted: Boolean = !Request.isValidBrokerId(replicaId)

    // 读取指定位置和大小的消息数据
    val logReadResults = this.readFromLocalLog(
        replicaId = replicaId,
        fetchOnlyFromLeader = fetchOnlyFromLeader,
        readOnlyCommitted = fetchOnlyCommitted,
        fetchMaxBytes = fetchMaxBytes,
        hardMaxBytesLimit = hardMaxBytesLimit,
        readPartitionInfo = fetchInfos,
        quota = quota)

    // 如果当前是来自 follower 的同步消息数据请求，则更新 follower 副本的状态，
    // 并尝试扩张 ISR 集合，同时尝试触发监听对应 topic 分区的 DelayedProduce 延时任务
    if (Request.isValidBrokerId(replicaId))
        this.updateFollowerLogReadResults(replicaId, logReadResults)

    val logReadResultValues = logReadResults.map { case (_, v) => v }
    val bytesReadable = logReadResultValues.map(_.info.records.sizeInBytes).sum
    val errorReadingData = logReadResultValues.foldLeft(false)(
        (errorIncurred, readResult) => errorIncurred || (readResult.error != Errors.NONE))

    if (timeout <= 0 // 请求希望立即响应
            || fetchInfos.isEmpty // 请求不期望有响应数据
            || bytesReadable >= fetchMinBytes // 已经有足够的数据可以响应
            || errorReadingData) { // 读取数据出现错误
        val fetchPartitionData = logReadResults.map { case (tp, result) =>
            tp -> FetchPartitionData(result.error, result.hw, result.info.records)
        }
        // 立即响应
        responseCallback(fetchPartitionData)
    } else {
        // 构造 DelayedFetch 延时任务
        val fetchPartitionStatus = logReadResults.map { case (topicPartition, result) =>
            val fetchInfo = fetchInfos.collectFirst {
                case (tp, v) if tp == topicPartition => v
            }.getOrElse(sys.error(s"Partition $topicPartition not found in fetchInfos"))
            (topicPartition, FetchPartitionStatus(result.info.fetchOffsetMetadata, fetchInfo))
        }
        val fetchMetadata = FetchMetadata(fetchMinBytes, fetchMaxBytes, hardMaxBytesLimit,
            fetchOnlyFromLeader, fetchOnlyCommitted, isFromFollower, replicaId, fetchPartitionStatus)
        val delayedFetch = new DelayedFetch(timeout, fetchMetadata, this, quota, responseCallback)

        // 构造延时任务关注的 key，即相应的 topic 分区对象
        val delayedFetchKeys = fetchPartitionStatus.map { case (tp, _) => new TopicPartitionOperationKey(tp) }

        // 交由炼狱管理
        delayedFetchPurgatory.tryCompleteElseWatch(delayedFetch, delayedFetchKeys)
    }
}

def readFromLocalLog(replicaId: Int, // 请求的 follower 副本 ID
                     fetchOnlyFromLeader: Boolean, // 是否只读 leader 副本的消息，一般 debug 模式下可以读 follower 副本的数据
                     readOnlyCommitted: Boolean, // 是否只读已完成提交的消息（即 HW 之前的消息），如果是来自消费者的请求则该参数是 true，如果是 follower 则该参数是 false
                     fetchMaxBytes: Int, // 最大 fetch 字节数
                     hardMaxBytesLimit: Boolean,
                     readPartitionInfo: Seq[(TopicPartition, PartitionData)], // 每个分区读取的起始 offset 和最大字节数
                     quota: ReplicaQuota): Seq[(TopicPartition, LogReadResult)] = {

    def read(tp: TopicPartition, fetchInfo: PartitionData, limitBytes: Int, minOneMessage: Boolean): LogReadResult = {
        val offset = fetchInfo.offset
        val partitionFetchSize = fetchInfo.maxBytes
        try {
            // 获取待读取消息的副本对象，一般都是从本地副本读取（debug 模式除外）
            val localReplica = if (fetchOnlyFromLeader) getLeaderReplicaIfLocal(tp) else getReplicaOrException(tp)
            // 计算读取消息的 offset 上界，如果是来自消费者的请求，则上界为 HW，如果是来自 follower 的请求，则上界为 LEO
            val maxOffsetOpt = if (readOnlyCommitted) Some(localReplica.highWatermark.messageOffset) else None
            val initialLogEndOffset = localReplica.logEndOffset.messageOffset // LEO
            val initialHighWatermark = localReplica.highWatermark.messageOffset // HW
            val fetchTimeMs = time.milliseconds
            val logReadInfo = localReplica.log match {
                case Some(log) =>
                    val adjustedFetchSize = math.min(partitionFetchSize, limitBytes)
                    // 从 Log 中读取消息数据
                    val fetch = log.read(offset, adjustedFetchSize, maxOffsetOpt, minOneMessage)

                    // 限流检测
                    if (shouldLeaderThrottle(quota, tp, replicaId)) FetchDataInfo(fetch.fetchOffsetMetadata, MemoryRecords.EMPTY)
                    // For FetchRequest version 3, we replace incomplete message sets with an empty one as consumers can make
                    // progress in such cases and don't need to report a `RecordTooLargeException`
                    else if (!hardMaxBytesLimit && fetch.firstEntryIncomplete) FetchDataInfo(fetch.fetchOffsetMetadata, MemoryRecords.EMPTY)
                    else fetch

                // 对应副本的 Log 对象不存在
                case None =>
                    error(s"Leader for partition $tp does not have a local log")
                    FetchDataInfo(LogOffsetMetadata.UnknownOffsetMetadata, MemoryRecords.EMPTY)
            }

            // 封装结果返回
            LogReadResult(info = logReadInfo,
                hw = initialHighWatermark,
                leaderLogEndOffset = initialLogEndOffset,
                fetchTimeMs = fetchTimeMs,
                readSize = partitionFetchSize,
                exception = None)
        } catch {
            // ... 省略异常处理
        }
    } // ~ end of read

    var limitBytes = fetchMaxBytes
    val result = new mutable.ArrayBuffer[(TopicPartition, LogReadResult)]
    var minOneMessage = !hardMaxBytesLimit
    // 遍历读取每个 topic 分区的消息数据
    readPartitionInfo.foreach {
        case (tp, fetchInfo) =>
            val readResult = read(tp, fetchInfo, limitBytes, minOneMessage)
            val messageSetSize = readResult.info.records.sizeInBytes
            if (messageSetSize > 0) minOneMessage = false
            limitBytes = math.max(0, limitBytes - messageSetSize)
            result += (tp -> readResult)
    }
    result
}

private def updateFollowerLogReadResults(replicaId: Int, readResults: Seq[(TopicPartition, LogReadResult)]) {
    debug("Recording follower broker %d log read results: %s ".format(replicaId, readResults))
    // 遍历处理对应 topic 分区的日志数据读取结果
    readResults.foreach {
        case (topicPartition, readResult) =>
            getPartition(topicPartition) match {
                case Some(partition) =>
                    // 更新指定 follower 副本的状态，并尝试扩张对应分区的 ISR 集合，以及后移 leader 副本的 HW 值
                    partition.updateReplicaLogReadResult(replicaId, readResult)
                    // 尝试执行 DelayedProduce 延时任务，因为此时对应 topic 分区下已经有新的消息成功写入
                    this.tryCompleteDelayedProduce(new TopicPartitionOperationKey(topicPartition))
                case None =>
                    warn("While recording the replica LEO, the partition %s hasn't been created.".format(topicPartition))
            }
    }
}

private[server] class MetadataCache(brokerId: Int) extends Logging {

    /** 缓存每个分区的状态信息 */
    private val cache = mutable.Map[String, mutable.Map[Int, PartitionStateInfo]]() // [topic, [分区 ID, 分区状态信息]]
    /** kafka controller leader 的 ID */
    private var controllerId: Option[Int] = None
    /** 记录当前可用的 broker 信息 */
    private val aliveBrokers = mutable.Map[Int, Broker]()
    /** 记录当前可用的 broker 节点信息 */
    private val aliveNodes = mutable.Map[Int, collection.Map[ListenerName, Node]]()

    // ... 省略方法定义

}

def updateCache(correlationId: Int, updateMetadataRequest: UpdateMetadataRequest): Seq[TopicPartition] = {
    inWriteLock(partitionMetadataLock) {
        // 更新本地缓存的 kafka controller 的 ID
        controllerId = updateMetadataRequest.controllerId match {
            case id if id < 0 => None
            case id => Some(id)
        }

        // 清除本地缓存的集群可用的 broker 节点信息，并由 UpdateMetadataRequest 请求重新构建
        aliveNodes.clear()
        aliveBrokers.clear()
        updateMetadataRequest.liveBrokers.asScala.foreach { broker =>
            // aliveNodes 是一个请求热点，所以这里使用 java.util.HashMap 来提升性能，如果是 scala 2.10 之后可以使用 AnyRefMap 代替
            val nodes = new java.util.HashMap[ListenerName, Node]
            val endPoints = new mutable.ArrayBuffer[EndPoint]
            broker.endPoints.asScala.foreach { ep =>
                endPoints += EndPoint(ep.host, ep.port, ep.listenerName, ep.securityProtocol)
                nodes.put(ep.listenerName, new Node(broker.id, ep.host, ep.port))
            }
            aliveBrokers(broker.id) = Broker(broker.id, endPoints, Option(broker.rack))
            aliveNodes(broker.id) = nodes.asScala
        }

        // 基于 UpdateMetadataRequest 请求更新每个分区的状态信息，并返回需要被移除的分区集合
        val deletedPartitions = new mutable.ArrayBuffer[TopicPartition]
        updateMetadataRequest.partitionStates.asScala.foreach {
            case (tp, info) =>
                val controllerId = updateMetadataRequest.controllerId
                val controllerEpoch = updateMetadataRequest.controllerEpoch
                // 如果请求标记对应的 topic 分区需要被删除
                if (info.leader == LeaderAndIsr.LeaderDuringDelete) {
                    // 删除本地缓存的对应 topic 分区的状态信息
                    this.removePartitionInfo(tp.topic, tp.partition)
                    deletedPartitions += tp
                } else {
                    // PartitionState -> PartitionStateInfo
                    val partitionInfo = this.partitionStateToPartitionStateInfo(info)
                    // 更新本地缓存的对应 topic 分区的状态信息
                    this.addOrUpdatePartitionInfo(tp.topic, tp.partition, partitionInfo)
                }
        }
        deletedPartitions
    }
}

def getTopicMetadata(topics: Set[String],
                     listenerName: ListenerName,
                     errorUnavailableEndpoints: Boolean = false): Seq[MetadataResponse.TopicMetadata] = {
    inReadLock(partitionMetadataLock) {
        topics.toSeq.flatMap { topic =>
            // 获取指定 topic 下分区元数据信息，并与 topic 一起构造 topic 元数据对象返回
            this.getPartitionMetadata(topic, listenerName, errorUnavailableEndpoints).map { partitionMetadata =>
                new MetadataResponse.TopicMetadata(Errors.NONE, topic, Topic.isInternal(topic), partitionMetadata.toBuffer.asJava)
            }
        }
    }
}

private def getPartitionMetadata(
                                 topic: String,
                                 listenerName: ListenerName,
                                 errorUnavailableEndpoints: Boolean): Option[Iterable[MetadataResponse.PartitionMetadata]] = {
    // 遍历每个 topic 对应的分区集合
    cache.get(topic).map { partitions =>
        partitions.map { case (partitionId, partitionState) =>
            val topicPartition = TopicAndPartition(topic, partitionId)

            // 获取分区对应的 LeaderAndIsr 对象，其中封装了对应分区的 leader 副本 ID 和 ISR 集合等信息
            val leaderAndIsr = partitionState.leaderIsrAndControllerEpoch.leaderAndIsr
            // 获取 leader 副本所在的节点信息
            val maybeLeader = this.getAliveEndpoint(leaderAndIsr.leader, listenerName)
            // 获取分区的 AR 集合
            val replicas = partitionState.allReplicas
            // 获取 AR 集合中可用的副本对应的节点信息
            val replicaInfo = this.getEndpoints(replicas, listenerName, errorUnavailableEndpoints)

            maybeLeader match {
                // 分区 leader 副本不可用
                case None =>
                    new MetadataResponse.PartitionMetadata(Errors.LEADER_NOT_AVAILABLE,
                        partitionId, Node.noNode(), replicaInfo.asJava, java.util.Collections.emptyList())
                case Some(leader) =>
                    // 获取分区的 ISR 集合
                    val isr = leaderAndIsr.isr
                    // 获取 ISR 集合中可用的副本对应的节点信息
                    val isrInfo = this.getEndpoints(isr, listenerName, errorUnavailableEndpoints)
                    if (replicaInfo.size < replicas.size) {
                        // 如果 AR 集合中存在不可用的副本，则返回 REPLICA_NOT_AVAILABLE 错误
                        new MetadataResponse.PartitionMetadata(Errors.REPLICA_NOT_AVAILABLE, partitionId, leader, replicaInfo.asJava, isrInfo.asJava)
                    } else if (isrInfo.size < isr.size) {
                        // 如果 ISR 集合中存在不可用的的副本，则返回 REPLICA_NOT_AVAILABLE 错误
                        new MetadataResponse.PartitionMetadata(Errors.REPLICA_NOT_AVAILABLE, partitionId, leader, replicaInfo.asJava, isrInfo.asJava)
                    } else {
                        // AR 集合和 ISR 集合中的副本都是可用的
                        new MetadataResponse.PartitionMetadata(Errors.NONE, partitionId, leader, replicaInfo.asJava, isrInfo.asJava)
                    }
            }
        }
    }
}