深入理解Kafka服务端之副本管理器介绍及Leader副本和Follower副本的切换

class ReplicaManager(val config: KafkaConfig,//配置管理类
                     metrics: Metrics,//监控指标类
                     time: Time,
                     val zkClient: KafkaZkClient,//zk客户端对象
                     scheduler: Scheduler,//调度器
                     val logManager: LogManager,//日志管理器
                     val isShuttingDown: AtomicBoolean,//是否已经关闭
                     quotaManagers: QuotaManagers,//配额管理器
                     val brokerTopicStats: BrokerTopicStats,
                     val metadataCache: MetadataCache,//Broker元数据缓存
                     logDirFailureChannel: LogDirFailureChannel,//失效日志路径的处理类
                     // 处理延时PRODUCE生产请求的Purgatory
                     val delayedProducePurgatory: DelayedOperationPurgatory[DelayedProduce],
                     // 处理延时FETCH拉取请求的Purgatory
                     val delayedFetchPurgatory: DelayedOperationPurgatory[DelayedFetch],
                     // 处理延时DELETE_RECORDS删除请求的Purgatory
                     val delayedDeleteRecordsPurgatory: DelayedOperationPurgatory[DelayedDeleteRecords],
                     // 处理延时ELECT_LEADERS leader选举请求的Purgatory
                     val delayedElectPreferredLeaderPurgatory: DelayedOperationPurgatory[DelayedElectPreferredLeader],
                     threadNamePrefix: Option[String]) extends Logging with KafkaMetricsGroup {
    ...
}

override def createFetcherThread(fetcherId: Int, sourceBroker: BrokerEndPoint): ReplicaFetcherThread = {
    val prefix = threadNamePrefix.map(tp => s"$tp:").getOrElse("")
    val threadName = s"${prefix}ReplicaFetcherThread-$fetcherId-${sourceBroker.id}"
    //创建ReplicaFetcherThread线程实例对象并返回
    new ReplicaFetcherThread(threadName, fetcherId, sourceBroker, brokerConfig, failedPartitions, replicaManager,
      metrics, time, quotaManager)
}

private def delayedProduceRequestRequired(requiredAcks: Short,
                                            entriesPerPartition: Map[TopicPartition, MemoryRecords],
                                            localProduceResults: Map[TopicPartition, LogAppendResult]): Boolean = {
    requiredAcks == -1 &&
    entriesPerPartition.nonEmpty &&
    localProduceResults.values.count(_.exception.isDefined) < entriesPerPartition.size
}

val produceMetadata = ProduceMetadata(requiredAcks, produceStatus)
//创建DelayedProduce延时请求对象
val delayedProduce = new DelayedProduce(timeout, produceMetadata, this, responseCallback, delayedProduceLock)
val producerRequestKeys = entriesPerPartition.keys.map(new TopicPartitionOperationKey(_)).toSeq
/再一次尝试完成该延时请求
//如果暂时无法完成，则将对象放入到相应的Purgatory中等待后续处理
delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys)

def fetchMessages(timeout: Long,//请求处理的超时时间
                    replicaId: Int,//副本ID。对于消费者来说，这个值为-1，如果是Follower副本，该值就是Follower副本所在节点的Broker ID
                    fetchMinBytes: Int,//能够获取的最小字节数
                    fetchMaxBytes: Int,//能够获取的最大字节数
                    hardMaxBytesLimit: Boolean,//是否能超过最大字节数限制，如果为True，则表示返回的数据字节数绝对不允许超过最大字节数
                    fetchInfos: Seq[(TopicPartition, PartitionData)],//读取分区的信息，比如读哪些分区，从分区的哪个偏移量开始读取，最多读取多少字节等
                    quota: ReplicaQuota = UnboundedQuota,//配额控制，判断是否需要在读取过程中做限速控制
                    responseCallback: Seq[(TopicPartition, FetchPartitionData)] => Unit,//回调逻辑
                    isolationLevel: IsolationLevel) {  
  ...
  if (timeout <= 0 || fetchInfos.isEmpty || bytesReadable >= fetchMinBytes || errorReadingData) {
        //如果满足一个条件，则返回
        val fetchPartitionData = logReadResults.map { case (tp, result) =>
          tp -> FetchPartitionData(result.error, result.highWatermark, result.leaderLogStartOffset, result.info.records,
            result.lastStableOffset, result.info.abortedTransactions)
        }
        //调用回调函数
        responseCallback(fetchPartitionData)


        //如果不满足上面4个条件，那么拉取请求会进行等待
      } else {


        val fetchPartitionStatus = new mutable.ArrayBuffer[(TopicPartition, FetchPartitionStatus)]
        fetchInfos.foreach { case (topicPartition, partitionData) =>
          logReadResultMap.get(topicPartition).foreach(logReadResult => {
            val logOffsetMetadata = logReadResult.info.fetchOffsetMetadata
            fetchPartitionStatus += (topicPartition -> FetchPartitionStatus(logOffsetMetadata, partitionData))
          })
        }
        val fetchMetadata = FetchMetadata(fetchMinBytes, fetchMaxBytes, hardMaxBytesLimit, fetchOnlyFromLeader,
          fetchIsolation, isFromFollower, replicaId, fetchPartitionStatus)
        //构建DelayedFetch延时请求对象
        val delayedFetch = new DelayedFetch(timeout, fetchMetadata, this, quota, responseCallback)


        val delayedFetchKeys = fetchPartitionStatus.map { case (tp, _) => new TopicPartitionOperationKey(tp) }


        //再次尝试完成请求，如果依然不能完成，则交由Purgatory等待后续处理
        delayedFetchPurgatory.tryCompleteElseWatch(delayedFetch, delayedFetchKeys)
  }
}

def becomeLeaderOrFollower(correlationId: Int,
                             leaderAndIsrRequest: LeaderAndIsrRequest,
                             onLeadershipChange: (Iterable[Partition], Iterable[Partition]) => Unit): LeaderAndIsrResponse = {
     ...
    replicaStateChangeLock synchronized {
      //如果leaderAndIsrRequest请求中携带的Controller Epoch小于当前节点的controllerEpoch，说明是过期Controller发送的请求
      if (leaderAndIsrRequest.controllerEpoch < controllerEpoch) {
        stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from controller ${leaderAndIsrRequest.controllerId} with " +
          s"correlation id $correlationId since its controller epoch ${leaderAndIsrRequest.controllerEpoch} is old. " +
          s"Latest known controller epoch is $controllerEpoch")
        //抛出响应的异常
        leaderAndIsrRequest.getErrorResponse(0, Errors.STALE_CONTROLLER_EPOCH.exception)
      } else {
        //初始化一个结果集，保存分区和对应处理结果
        val responseMap = new mutable.HashMap[TopicPartition, Errors]
        //获取控制器节点的BrokerId
        val controllerId = leaderAndIsrRequest.controllerId
        //更新当前的Controller Epoch 值
        controllerEpoch = leaderAndIsrRequest.controllerEpoch


        //定义存储分区信息的集合
        val partitionState = new mutable.HashMap[Partition, LeaderAndIsrRequest.PartitionState]()
        val updatedPartitions = new mutable.HashSet[Partition]
        // 遍历LeaderAndIsrRequest请求中的所有分区状态，进行Leader Epoch校验
        leaderAndIsrRequest.partitionStates.asScala.foreach { case (topicPartition, stateInfo) =>
          // 从allPartitions中获取对应分区对象,如果没有则创建
          val partition = getOrCreatePartition(topicPartition)
          //获取当前分区对象的Leader Epoch
          val currentLeaderEpoch = partition.getLeaderEpoch
          //获取请求中携带的Leader Epoch
          val requestLeaderEpoch = stateInfo.basePartitionState.leaderEpoch
          //如果是离线分区，将该分区放入结果集，并封装一个KAFKA_STORAGE_ERROR错误
          if (partition eq ReplicaManager.OfflinePartition) {
            stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from " +
              s"controller $controllerId with correlation id $correlationId " +
              s"epoch $controllerEpoch for partition $topicPartition as the local replica for the " +
              "partition is in an offline log directory")
            responseMap.put(topicPartition, Errors.KAFKA_STORAGE_ERROR)
          //如果请求中的Leader Epoch > 当前分区的Leader Epoch
          } else if (requestLeaderEpoch > currentLeaderEpoch) {
            //且分区的初始副本分配中包含当前节点
            if (stateInfo.basePartitionState.replicas.contains(localBrokerId)) {
              //将分区对象放入updatedPartitions集合
              updatedPartitions.add(partition)
              //将分区状态放入partitionState集合
              partitionState.put(partition, stateInfo)
            } else {
              stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from controller $controllerId with " +
                s"correlation id $correlationId epoch $controllerEpoch for partition $topicPartition as itself is not " +
                s"in assigned replica list ${stateInfo.basePartitionState.replicas.asScala.mkString(",")}")
              responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_OR_PARTITION)
            }
            //如果请求中的Leader Epoch < 当前分区的Leader Epoch，说明是过期leader的请求
          } else if (requestLeaderEpoch < currentLeaderEpoch) {
            stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from " +
              s"controller $controllerId with correlation id $correlationId " +
              s"epoch $controllerEpoch for partition $topicPartition since its associated " +
              s"leader epoch $requestLeaderEpoch is smaller than the current " +
              s"leader epoch $currentLeaderEpoch")
            //将该分区放入结果集，并封装一个STALE_CONTROLLER_EPOCH错误
            responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH)
          } else {
            stateChangeLogger.debug(s"Ignoring LeaderAndIsr request from " +
              s"controller $controllerId with correlation id $correlationId " +
              s"epoch $controllerEpoch for partition $topicPartition since its associated " +
              s"leader epoch $requestLeaderEpoch matches the current leader epoch")
            //将该分区放入结果集，并封装一个STALE_CONTROLLER_EPOCH错误
            responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH)
          }
        }
        // 确定Broker上副本是哪些分区的Leader副本
        val partitionsTobeLeader = partitionState.filter { case (_, stateInfo) =>
          stateInfo.basePartitionState.leader == localBrokerId
        }
        // 确定Broker上副本是哪些分区的Follower副本
        val partitionsToBeFollower = partitionState -- partitionsTobeLeader.keys


        //为partitionsTobeLeader集合中的所有分区在本机的副本执行成为Leader的操作，调用makeLeaders方法
        val partitionsBecomeLeader = if (partitionsTobeLeader.nonEmpty)
          makeLeaders(controllerId, controllerEpoch, partitionsTobeLeader, correlationId, responseMap)
        else
          Set.empty[Partition]
        //为partitionsToBeFollower集合中的所有分区在本机的副本执行成为Follower的操作，调用makeFollowers方法
        val partitionsBecomeFollower = if (partitionsToBeFollower.nonEmpty)
          makeFollowers(controllerId, controllerEpoch, partitionsToBeFollower, correlationId, responseMap)
        else
          Set.empty[Partition]


        //遍历请求中的所有分区
        leaderAndIsrRequest.partitionStates.asScala.keys.foreach { topicPartition =>
          /*
           * 如果指定了一个离线的分区路径，那么通过getOrCreatePartition方法创建分区的本地副本抛出异常之前，getOrCreatePartition方法
           * 可能已经创建了一个对应的分区对象了，这时在allPartitions对象池中，就会添加一个Partition对象为空的TopicPartition，
           * 这里将这个空的Partition对象修改为OfflinePartition
           */
          if (localReplica(topicPartition).isEmpty && (allPartitions.get(topicPartition) ne ReplicaManager.OfflinePartition))
            //如果当前节点没有指定分区的本地副本，且该分区的状态为离线状态，则更新该分区为离线分区
            allPartitions.put(topicPartition, ReplicaManager.OfflinePartition)
        }


        if (!hwThreadInitialized) {
          // 启动高水位检查点专属线程
          // 定期将Broker上所有非Offline分区的高水位值写入到检查点文件,默认每5秒执行1次
          startHighWaterMarksCheckPointThread()
          hwThreadInitialized = true
        }


        val futureReplicasAndInitialOffset = new mutable.HashMap[TopicPartition, InitialFetchState]
        //遍历前面Leader Epoch校验正常的分区，看是否需要做副本迁移
        for (partition <- updatedPartitions) {
          val topicPartition = partition.topicPartition
          //如果已经定义了指定分区对应的Future日志对象
          // 这里用于做副本迁移，当用户希望将副本从一个目录迁移到另一个目录时，会将迁移后的目录定义为-future后缀的目录
          //然后在该目录下创建future日志，当future日志追上current日志时，来替换当前分区的日志
          if (logManager.getLog(topicPartition, isFuture = true).isDefined) {
            partition.localReplica.foreach { replica =>
              val leader = BrokerEndPoint(config.brokerId, "localhost", -1)


              partition.getOrCreateReplica(Request.FutureLocalReplicaId, isNew = false)


              logManager.abortAndPauseCleaning(topicPartition)


              futureReplicasAndInitialOffset.put(topicPartition, InitialFetchState(leader,
                partition.getLeaderEpoch, replica.highWatermark.messageOffset))
            }
          }
        }
        //副本修改日志路径管理器拉取指定分区的数据到future副本
        replicaAlterLogDirsManager.addFetcherForPartitions(futureReplicasAndInitialOffset)
        // 关闭空闲副本拉取线程
        replicaFetcherManager.shutdownIdleFetcherThreads()
        // 关闭空闲日志路径数据迁移线程
        replicaAlterLogDirsManager.shutdownIdleFetcherThreads()
        // 执行Leader变更之后的回调逻辑
        onLeadershipChange(partitionsBecomeLeader, partitionsBecomeFollower)
        //封装LeaderAndIsrResponse响应
        new LeaderAndIsrResponse(Errors.NONE, responseMap.asJava)
      }
    }
}

if (leaderAndIsrRequest.controllerEpoch < controllerEpoch) {
    ...
    //抛出响应的异常
    leaderAndIsrRequest.getErrorResponse(0, Errors.STALE_CONTROLLER_EPOCH.exception)
}

//初始化一个结果集，保存分区和对应处理结果
val responseMap = new mutable.HashMap[TopicPartition, Errors]

//获取控制器节点的BrokerId
val controllerId = leaderAndIsrRequest.controllerId
//更新当前的Controller Epoch 值
controllerEpoch = leaderAndIsrRequest.controllerEpoch

//定义存储分区信息的集合
val partitionState = new mutable.HashMap[Partition, LeaderAndIsrRequest.PartitionState]()
//定义Leader Epoch校验通过的分区集合
val updatedPartitions = new mutable.HashSet[Partition]

// 遍历LeaderAndIsrRequest请求中的所有分区状态，进行Leader Epoch校验
leaderAndIsrRequest.partitionStates.asScala.foreach { case (topicPartition, stateInfo) =>
    // 从allPartitions中获取对应分区对象,如果没有则创建
    val partition = getOrCreatePartition(topicPartition)
    //获取当前分区对象的Leader Epoch
    val currentLeaderEpoch = partition.getLeaderEpoch
    //获取请求中携带的Leader Epoch
    val requestLeaderEpoch = stateInfo.basePartitionState.leaderEpoch
    //如果是离线分区，将该分区放入结果集，并封装一个KAFKA_STORAGE_ERROR错误
    if (partition eq ReplicaManager.OfflinePartition) {
        ...
        responseMap.put(topicPartition, Errors.KAFKA_STORAGE_ERROR)
        //如果请求中的Leader Epoch > 当前分区的Leader Epoch
    } else if (requestLeaderEpoch > currentLeaderEpoch) {
        //且分区的初始副本分配中包含当前节点
        if (stateInfo.basePartitionState.replicas.contains(localBrokerId)) {
            //将分区对象放入updatedPartitions集合
            updatedPartitions.add(partition)
            //将分区状态放入partitionState集合
            partitionState.put(partition, stateInfo)
        } else {
            ...
            responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_OR_PARTITION)
        }
        //如果请求中的Leader Epoch < 当前分区的Leader Epoch，说明是过期leader的请求
    } else if (requestLeaderEpoch < currentLeaderEpoch) {
        ...
        //将该分区放入结果集，并封装一个STALE_CONTROLLER_EPOCH错误
        responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH)
    } else {
        ...
        //将该分区放入结果集，并封装一个STALE_CONTROLLER_EPOCH错误
        responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH)
    }
}

// 确定Broker上的本地副本是哪些分区的Leader副本
val partitionsTobeLeader = partitionState.filter { case (_, stateInfo) =>
  stateInfo.basePartitionState.leader == localBrokerId
}

// 确定Broker上的本地副本是哪些分区的Follower副本
val partitionsToBeFollower = partitionState -- partitionsTobeLeader.keys

//为partitionsTobeLeader集合中的所有分区在本机的副本执行成为Leader的操作，调用makeLeaders方法
val partitionsBecomeLeader = if (partitionsTobeLeader.nonEmpty)
  makeLeaders(controllerId, controllerEpoch, partitionsTobeLeader, correlationId, responseMap)
else
  Set.empty[Partition]

//为partitionsToBeFollower集合中的所有分区在本机的副本执行成为Follower的操作，调用makeFollowers方法
val partitionsBecomeFollower = if (partitionsToBeFollower.nonEmpty)
  makeFollowers(controllerId, controllerEpoch, partitionsToBeFollower, correlationId, responseMap)
else
  Set.empty[Partition]

private def makeLeaders(controllerId: Int,//控制器所在节点ID
                          epoch: Int,//控制器纪元值
                          partitionState: Map[Partition, LeaderAndIsrRequest.PartitionState],//要成为Leader副本的分区信息集合
                          correlationId: Int,//
                          responseMap: mutable.Map[TopicPartition, Errors]//按照分区分组的错误信息
                         ): Set[Partition] = {
    partitionState.keys.foreach { partition =>
      ...
    }
    //遍历要成为Leader副本的分区信息集合
    for (partition <- partitionState.keys)
      //初始化分区结果，给每个分区添加Errors.NONE信息
      responseMap.put(partition.topicPartition, Errors.NONE)
    //定义一个集合用于存储本地副本成为Leader副本的分区对象
    val partitionsToMakeLeaders = mutable.Set[Partition]()


    try {
      //停止本地副本请要成为Leader副本的所有分区的数据拉取
      replicaFetcherManager.removeFetcherForPartitions(partitionState.keySet.map(_.topicPartition))
      //遍历更新所有要成为Leader副本的分区的分区信息
      partitionState.foreach{ case (partition, partitionStateInfo) =>
        try {
          //判断指定分区在本地的副本之前是否为Leader副本，如果之前就是Leader副本，makeLeader方法返回false，否则返回true，执行分支内容
          if (partition.makeLeader(controllerId, partitionStateInfo, correlationId)) {
            //将该副本对象放入集合
            partitionsToMakeLeaders += partition
            ...
          } else
            ...
        } catch {
          case e: KafkaStorageException =>
            ...
            val dirOpt = getLogDir(partition.topicPartition)
            error(s"Error while making broker the leader for partition $partition in dir $dirOpt", e)
            // 把KAFKA_SOTRAGE_ERRROR异常封装到Response中
            responseMap.put(partition.topicPartition, Errors.KAFKA_STORAGE_ERROR)
        }
      }


    } catch {
      case e: Throwable =>
        ...
        throw e
    }


    partitionState.keys.foreach { partition =>
      ...
    }


    partitionsToMakeLeaders
  }

def makeLeader(controllerId: Int, partitionStateInfo: LeaderAndIsrRequest.PartitionState, correlationId: Int): Boolean = {
    val (leaderHWIncremented, isNewLeader) = inWriteLock(leaderIsrUpdateLock) {
      //获取控制器发来的分区副本所在节点的集合
      val newAssignedReplicas = partitionStateInfo.basePartitionState.replicas.asScala.map(_.toInt)
      //更新controller epoch
      controllerEpoch = partitionStateInfo.basePartitionState.controllerEpoch
      // add replicas that are new
      //根据控制器发来的ISR列表获取对应的副本对象集合
      val newInSyncReplicas = partitionStateInfo.basePartitionState.isr.asScala.map(r => getOrCreateReplica(r, partitionStateInfo.isNew)).toSet
      //移除本地保存的不在控制器发来的ISR列表中的副本对象
      (assignedReplicas.map(_.brokerId) -- newAssignedReplicas).foreach(removeReplica)
      //更新本地 ISR 列表
      inSyncReplicas = newInSyncReplicas


      newAssignedReplicas.foreach(id => getOrCreateReplica(id, partitionStateInfo.isNew))
      //获取本地副本对象
      val leaderReplica = localReplicaOrException
      //将本地副本对象的LEO作为新的Leader Epoch的StartOffset
      val leaderEpochStartOffset = leaderReplica.logEndOffset
      info(s"$topicPartition starts at Leader Epoch ${partitionStateInfo.basePartitionState.leaderEpoch} from " +
        s"offset $leaderEpochStartOffset. Previous Leader Epoch was: $leaderEpoch")


      //更新Leader Epoch
      leaderEpoch = partitionStateInfo.basePartitionState.leaderEpoch
      //更新Leader Epoch 对应的startOffset
      leaderEpochStartOffsetOpt = Some(leaderEpochStartOffset)
      //更新zkVersion
      zkVersion = partitionStateInfo.basePartitionState.zkVersion


      // 尝试更新Leader Epoch和对应的StartOffset
      leaderReplica.log.foreach { log =>
        log.maybeAssignEpochStartOffset(leaderEpoch, leaderEpochStartOffset)
      }
      //标记将要称为Leader的副本是否是新转换成Leader副本的
      //如果leaderReplicaIdOpt为当前节点id，说明不是新转换成的，该分区之前的Leader副本就在当前节点上
      val isNewLeader = !leaderReplicaIdOpt.contains(localBrokerId)
      //获取当前Leader副本的LEO
      val curLeaderLogEndOffset = leaderReplica.logEndOffset
      val curTimeMs = time.milliseconds
      //遍历当前节点保存的所有的Follower副本对象
      (assignedReplicas - leaderReplica).foreach { replica =>


        val lastCaughtUpTimeMs = if (inSyncReplicas.contains(replica)) curTimeMs else 0L
        //更新所有Follower副本对象的_lastCaughtUpTimeMs/lastFetchLeaderLogEndOffset/lastFetchTimeMs
        replica.resetLastCaughtUpTime(curLeaderLogEndOffset, curTimeMs, lastCaughtUpTimeMs)
      }


      if (isNewLeader) {
        //定义Leader副本的HW
        leaderReplica.convertHWToLocalOffsetMetadata()
        //更新Leader副本的id为当前broker的id
        leaderReplicaIdOpt = Some(localBrokerId)
        //更新本地存储的所有Follower副本的LEO
        assignedReplicas.filter(_.brokerId != localBrokerId).foreach(_.updateLogReadResult(LogReadResult.UnknownLogReadResult))
      }
      (maybeIncrementLeaderHW(leaderReplica), isNewLeader)
    }
    //如果HW修改了，尝试执行之前延迟处理的请求
    if (leaderHWIncremented)
      tryCompleteDelayedRequests()
    isNewLeader
}

val newAssignedReplicas = partitionStateInfo.basePartitionState.replicas.asScala.map(_.toInt)

controllerEpoch = partitionStateInfo.basePartitionState.controllerEpoch

val newInSyncReplicas = partitionStateInfo.basePartitionState.isr.asScala.map(r => getOrCreateReplica(r, partitionStateInfo.isNew)).toSet

(assignedReplicas.map(_.brokerId) -- newAssignedReplicas).foreach(removeReplica)

inSyncReplicas = newInSyncReplicas

newAssignedReplicas.foreach(id => getOrCreateReplica(id, partitionStateInfo.isNew))

//获取本地副本对象
val leaderReplica = localReplicaOrException
//将本地副本对象的LEO作为新的Leader Epoch的StartOffset
val leaderEpochStartOffset = leaderReplica.logEndOffset

//更新Leader Epoch
leaderEpoch = partitionStateInfo.basePartitionState.leaderEpoch
//更新Leader Epoch 对应的startOffset
leaderEpochStartOffsetOpt = Some(leaderEpochStartOffset)
//更新zkVersion
zkVersion = partitionStateInfo.basePartitionState.zkVersion


// 尝试更新Log日志对象的Leader Epoch和对应的StartOffset
leaderReplica.log.foreach { log =>
  log.maybeAssignEpochStartOffset(leaderEpoch, leaderEpochStartOffset)
}

//标记将要称为Leader的副本是否是新转换成Leader副本的
//如果leaderReplicaIdOpt为当前节点id，说明不是新转换成的，该分区之前的Leader副本就在当前节点上
val isNewLeader = !leaderReplicaIdOpt.contains(localBrokerId)

//获取当前Leader副本的LEO
val curLeaderLogEndOffset = leaderReplica.logEndOffset
val curTimeMs = time.milliseconds
//遍历当前节点保存的所有的Follower副本对象
(assignedReplicas - leaderReplica).foreach { replica =>


  val lastCaughtUpTimeMs = if (inSyncReplicas.contains(replica)) curTimeMs else 0L
  //更新所有Follower副本对象的_lastCaughtUpTimeMs/lastFetchLeaderLogEndOffset/lastFetchTimeMs
  replica.resetLastCaughtUpTime(curLeaderLogEndOffset, curTimeMs, lastCaughtUpTimeMs)
}

if (isNewLeader) {
    //定义Leader副本的HW
    leaderReplica.convertHWToLocalOffsetMetadata()
    // mark local replica as the leader after converting hw
    //更新Leader副本的id为当前broker的id
    leaderReplicaIdOpt = Some(localBrokerId)
    // reset log end offset for remote replicas
    //更新本地存储的所有Follower副本的LEO
    assignedReplicas.filter(_.brokerId != localBrokerId).foreach(_.updateLogReadResult(LogReadResult.UnknownLogReadResult))
}

if (leaderHWIncremented)
  tryCompleteDelayedRequests()

private def makeFollowers(controllerId: Int,
                            epoch: Int,
                            partitionStates: Map[Partition, LeaderAndIsrRequest.PartitionState],
                            correlationId: Int,
                            responseMap: mutable.Map[TopicPartition, Errors]) : Set[Partition] = {
    partitionStates.foreach { case (partition, partitionState) =>
      ...
    }
    //遍历所有分区，将分区的异常类型初始化为Errors.NONE
    for (partition <- partitionStates.keys)
      responseMap.put(partition.topicPartition, Errors.NONE)
    //定义一个集合，用于存放在当前节点上本地副本变更为Follower副本的分区
    val partitionsToMakeFollower: mutable.Set[Partition] = mutable.Set()


    try {
      partitionStates.foreach { case (partition, partitionStateInfo) =>
          //获取控制器发来的分区Leader副本的brokerId
        val newLeaderBrokerId = partitionStateInfo.basePartitionState.leader
        try {
          metadataCache.getAliveBrokers.find(_.id == newLeaderBrokerId) match {
            //在元数据缓存中找到Leader副本的节点信息
            case Some(_) =>
              //如果给定分区的本地副本是转换成Follower副本的，makeFollower返回true，并执行一些列操作
              if (partition.makeFollower(controllerId, partitionStateInfo, correlationId))
                //将副本对象添加到本地副本新转换成Follower副本的集合
                partitionsToMakeFollower += partition
              else
               ...
            //如果元数据缓存中没有指定Leader副本所在节点的信息，说明Leader副本所在的节点不可用
            case None =>
              ...
              //创建一个本地副本对象，初始化日志对象
              partition.getOrCreateReplica(localBrokerId, isNew = partitionStateInfo.isNew)
          }
        } catch {
          case e: KafkaStorageException =>
            ...
            val dirOpt = getLogDir(partition.topicPartition)
            ...
            responseMap.put(partition.topicPartition, Errors.KAFKA_STORAGE_ERROR)
        }
      }
      //关闭现有的Fetcher线程
      replicaFetcherManager.removeFetcherForPartitions(partitionsToMakeFollower.map(_.topicPartition))
      partitionsToMakeFollower.foreach { partition =>
        ...
      }
      //遍历本地副本成为Follower副本的分区，尝试完成延迟请求
      partitionsToMakeFollower.foreach { partition =>
        val topicPartitionOperationKey = new TopicPartitionOperationKey(partition.topicPartition)
        tryCompleteDelayedProduce(topicPartitionOperationKey)
        tryCompleteDelayedFetch(topicPartitionOperationKey)
      }


      partitionsToMakeFollower.foreach { partition =>
       ...
      }


      if (isShuttingDown.get()) {
        partitionsToMakeFollower.foreach { partition =>
          ...
        }
      }
      else {
        val partitionsToMakeFollowerWithLeaderAndOffset = partitionsToMakeFollower.map { partition =>
          //找到分区Leader副本所在节点id
          val leader = metadataCache.getAliveBrokers.find(_.id == partition.leaderReplicaIdOpt.get).get
            .brokerEndPoint(config.interBrokerListenerName)
          //获取要拉取数据的起始偏移量
          val fetchOffset = partition.localReplicaOrException.highWatermark.messageOffset
          partition.topicPartition -> InitialFetchState(leader, partition.getLeaderEpoch, fetchOffset)
        }.toMap
        //TODO 添加新的拉取线程，让Follower Replica 去Leader Replica 拉取数据
        replicaFetcherManager.addFetcherForPartitions(partitionsToMakeFollowerWithLeaderAndOffset)


        partitionsToMakeFollower.foreach { partition =>
         ...
        }
      }
    } catch {
      case e: Throwable =>
        ...
        throw e
    }


    partitionStates.keys.foreach { partition =>
      ...
    }
    partitionsToMakeFollower
  }

def makeFollower(controllerId: Int, partitionStateInfo: LeaderAndIsrRequest.PartitionState, correlationId: Int): Boolean = {
  inWriteLock(leaderIsrUpdateLock) {
    //获取控制器发来的分区所有副本所在节点的brokerId集合
    val newAssignedReplicas = partitionStateInfo.basePartitionState.replicas.asScala.map(_.toInt)
    //获取分区Leader副本的节点id
    val newLeaderBrokerId = partitionStateInfo.basePartitionState.leader
    //获取当前分区的Leader Epoch
    val oldLeaderEpoch = leaderEpoch
    //更新controller epoch
    controllerEpoch = partitionStateInfo.basePartitionState.controllerEpoch
    //创建控制器分配的所有副本对象
    newAssignedReplicas.foreach(r => getOrCreateReplica(r, partitionStateInfo.isNew))
    //移除多余的副本对象(假设之前副本在0,1,2节点上，现在控制器要求副本在1,2,3节点上，那么就移除本地存储的节点0对应的副本对象)
    (assignedReplicas.map(_.brokerId) -- newAssignedReplicas).foreach(removeReplica)
    //清空ISR列表
    inSyncReplicas = Set.empty[Replica]
    //更新Leader Epoch
    leaderEpoch = partitionStateInfo.basePartitionState.leaderEpoch
    //更新Leader Epoch对应的StartOffset
    leaderEpochStartOffsetOpt = None
    //更新zkVersion
    zkVersion = partitionStateInfo.basePartitionState.zkVersion
    //如果控制器分配的Leader副本就是之前的Leader副本，且Leader Epoch没有改变，说明Leader副本没有切换，那么返回false
    // 否则更新leaderReplicaIdOpt信息，返回true
    if (leaderReplicaIdOpt.contains(newLeaderBrokerId) && leaderEpoch == oldLeaderEpoch) {
      false
    } else {
      leaderReplicaIdOpt = Some(newLeaderBrokerId)
      true
    }
  }
}

  if (!hwThreadInitialized) {
    // 启动高水位检查点线程
    // 定期将Broker上所有非Offline分区的高水位值写入到检查点文件,默认每5秒执行1次
    startHighWaterMarksCheckPointThread()
    hwThreadInitialized = true
  }


  val futureReplicasAndInitialOffset = new mutable.HashMap[TopicPartition, InitialFetchState]
  //遍历前面Leader Epoch校验正常的分区，看是否需要做副本迁移
  for (partition <- updatedPartitions) {
    val topicPartition = partition.topicPartition
    //如果已经定义了指定分区对应的Future日志对象
    // 这里用于做副本迁移，当用户希望将副本从一个目录迁移到另一个目录时，会将迁移后的目录定义为-future后缀的目录
    //然后在该目录下创建future日志，当future日志追上current日志时，来替换当前分区的日志
    if (logManager.getLog(topicPartition, isFuture = true).isDefined) {
      partition.localReplica.foreach { replica =>
        val leader = BrokerEndPoint(config.brokerId, "localhost", -1)
  
        //在分区的副本集合中添加Future副本
        partition.getOrCreateReplica(Request.FutureLocalReplicaId, isNew = false)
  
        //暂停分区日志的清理
        logManager.abortAndPauseCleaning(topicPartition)


        futureReplicasAndInitialOffset.put(topicPartition, InitialFetchState(leader,
          partition.getLeaderEpoch, replica.highWatermark.messageOffset))
      }
    }
  }
  //对指定的分区进行日志数据的迁移
  replicaAlterLogDirsManager.addFetcherForPartitions(futureReplicasAndInitialOffset)
  // 关闭空闲副本拉取线程
  replicaFetcherManager.shutdownIdleFetcherThreads()
  // 关闭空闲日志路径数据迁移线程
  replicaAlterLogDirsManager.shutdownIdleFetcherThreads()
  // 执行Leader变更之后的回调逻辑
  onLeadershipChange(partitionsBecomeLeader, partitionsBecomeFollower)
  //封装LeaderAndIsrResponse响应
  new LeaderAndIsrResponse(Errors.NONE, responseMap.asJava)
}