深入理解Kafka服务端之基于Leader Epoch的日志截断流程分析

override def doWork() {
  //尝试日志截断
  maybeTruncate()
  //尝试拉取数据
  maybeFetch()
}

private def maybeTruncate(): Unit = {
  // 将所有处于截断中状态的分区依据有无Leader Epoch值进行分组
  val (partitionsWithEpochs, partitionsWithoutEpochs) = fetchTruncatingPartitions()
  // 对于有Leader Epoch值的分区，将日志截断到Leader Epoch值对应的位移值处
  if (partitionsWithEpochs.nonEmpty) {
    truncateToEpochEndOffsets(partitionsWithEpochs)
  }
  // 对于没有Leader Epoch值的分区，将日志截断到高水位值处
  if (partitionsWithoutEpochs.nonEmpty) {
    truncateToHighWatermark(partitionsWithoutEpochs)
  }
}

private[server] def truncateToHighWatermark(partitions: Set[TopicPartition]): Unit = inLock(partitionMapLock) {


  val fetchOffsets = mutable.HashMap.empty[TopicPartition, OffsetTruncationState]
  // 遍历每个要执行截断操作的分区对象
  for (tp <- partitions) {
    // 获取分区的分区读取状态
    val partitionState = partitionStates.stateValue(tp)
    if (partitionState != null) {
      // 取出高水位值。
      val highWatermark = partitionState.fetchOffset
      val truncationState = OffsetTruncationState(highWatermark, truncationCompleted = true)


      info(s"Truncating partition $tp to local high watermark $highWatermark")
      // 执行截断到高水位值
      if (doTruncate(tp, truncationState))
        //保存分区和对应的截取状态
        fetchOffsets.put(tp, truncationState)
    }
  }
  // 更新这组分区的分区读取状态
  updateFetchOffsetAndMaybeMarkTruncationComplete(fetchOffsets)
}

private def truncateToEpochEndOffsets(latestEpochsForPartitions: Map[TopicPartition, EpochData]): Unit = {
  //返回一组[分区 -> epoch + endOffset]
  val endOffsets = fetchEpochEndOffsets(latestEpochsForPartitions)
  inLock(partitionMapLock) {
    //获取一组在向Leader副本请求截断位置过程中没有发生leader副本切换的[分区 -> epoch + endOffset]
    val epochEndOffsets = endOffsets.filter { case (tp, _) =>
      //获取当前的分区状态
      val curPartitionState = partitionStates.stateValue(tp)
      //根据分区获取对应的OffsetsForLeaderEpochRequest请求对象
      val partitionEpochRequest = latestEpochsForPartitions.get(tp).getOrElse {
        throw new IllegalStateException(
          s"Leader replied with partition $tp not requested in OffsetsForLeaderEpoch request")
      }
      //获取请求中携带的Follower副本的LeaderEpoch值
      val leaderEpochInRequest = partitionEpochRequest.currentLeaderEpoch.get
      //如果分区状态不为null，且请求携带的LeaderEpoch和当前分区状态中的LeaderEpoch一致，即发送请求的过程中没有发生leader副本的切换
      curPartitionState != null && leaderEpochInRequest == curPartitionState.currentLeaderEpoch
    }
    //调用maybeTruncateToEpochEndOffsets进行日志截断，结果包含两个对象：
    //fetchOffsets：进行正常截断的分区和对应的截断状态
    //partitionsWithError：出错的分区
    val ResultWithPartitions(fetchOffsets, partitionsWithError) = maybeTruncateToEpochEndOffsets(epochEndOffsets, latestEpochsForPartitions)
    //处理出错的分区，即将分区对象从集合头部移除，然后放到集合的末尾
    handlePartitionsWithErrors(partitionsWithError, "truncateToEpochEndOffsets")
    //更新分区的状态
    updateFetchOffsetAndMaybeMarkTruncationComplete(fetchOffsets)
  }
}

override def fetchEpochEndOffsets(partitions: Map[TopicPartition, EpochData]): Map[TopicPartition, EpochEndOffset] = {


  if (partitions.isEmpty) {
    debug("Skipping leaderEpoch request since all partitions do not have an epoch")
    return Map.empty
  }
  //封装OffsetsForLeaderEpochRequest请求
  val epochRequest = OffsetsForLeaderEpochRequest.Builder.forFollower(offsetForLeaderEpochRequestVersion, partitions.asJava, brokerConfig.brokerId)
  debug(s"Sending offset for leader epoch request $epochRequest")


  try {
    //向Leader副本所在节点发送OffsetsForLeaderEpochRequest请求并接受返回的OffsetsForLeaderEpochResponse响应
    val response = leaderEndpoint.sendRequest(epochRequest)
    //获取Leader返回的响应体
    val responseBody = response.responseBody.asInstanceOf[OffsetsForLeaderEpochResponse]
    debug(s"Received leaderEpoch response $response")
    //返回一组 [分区-> epoch + endOffset]
    responseBody.responses.asScala
  } catch {
    case t: Throwable =>
      warn(s"Error when sending leader epoch request for $partitions", t)
      val error = Errors.forException(t)
      partitions.map { case (tp, _) =>
        tp -> new EpochEndOffset(error, UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET)
      }
  }
}

//封装OffsetsForLeaderEpochRequest请求
val epochRequest = OffsetsForLeaderEpochRequest.Builder.forFollower(offsetForLeaderEpochRequestVersion, partitions.asJava, brokerConfig.brokerId)
//向Leader副本所在节点发送OffsetsForLeaderEpochRequest请求并接受返回的OffsetsForLeaderEpochResponse响应
val response = leaderEndpoint.sendRequest(epochRequest)

//处理OffsetsForLeaderEpochRequest
case ApiKeys.OFFSET_FOR_LEADER_EPOCH => handleOffsetForLeaderEpochRequest(request)

def endOffsetFor(requestedEpoch: Int): (Int, Long) = {
  inReadLock(lock) {
    val epochAndOffset =
      //如果请求的Leader Epoch = UNDEFINED_EPOCH
      if (requestedEpoch == UNDEFINED_EPOCH) {
      //返回的Leader Epoch 和 EndOffset为(UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET)
        (UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET)
        //如果请求的Leader Epoch 就是Leader副本当前的Leader Epoch，则返回LEO
      } else if (latestEpoch.contains(requestedEpoch)) {
      //返回(请求Epoch,Leader副本LEO)
        (requestedEpoch, logEndOffset())
      } else {
        //按照是否大于Follower请求的Epoch，将Leader端的Epoch缓存进行分组
        //subsequentEpochs：大于请求Epoch的Epochs集合
        //previousEpochs：小于等于请求Epoch的Epochs集合
        val (subsequentEpochs, previousEpochs) = epochs.partition { e => e.epoch > requestedEpoch}
        //如果Leader端缓存的Epoch都比请求的Epoch小
        if (subsequentEpochs.isEmpty) {
          //返回(UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET)
          (UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET)
        //如果Leader端缓存的Epoch不小于（≥）请求的Epoch
        } else if (previousEpochs.isEmpty) {
          //如果请求的epoch 比leader端保存的任何一个epoch都小，那么endOffset取最小epoch 对应的startOffset
          (requestedEpoch, subsequentEpochs.head.startOffset)
          //如果Leader端缓存的Epoch既有大于请求Epoch的，也有小于等于请求Epoch的，
        } else {
          // 针对Epoch：返回不大于请求Epoch的Epoch
          //针对endOffset：返回第一个大于请求Epoch的Epoch对应的StartOffset
          (previousEpochs.last.epoch, subsequentEpochs.head.startOffset)
        }
      }
    debug(s"Processed end offset request for epoch $requestedEpoch and returning epoch ${epochAndOffset._1} " +
      s"with end offset ${epochAndOffset._2} from epoch cache of size ${epochs.size}")
    epochAndOffset
  }
}

val epochEndOffsets = endOffsets.filter { case (tp, _) =>
    //获取当前的分区状态
    val curPartitionState = partitionStates.stateValue(tp)
    //根据分区获取 OffsetsForLeaderEpochRequest请求中的分区信息
    val partitionEpochRequest = latestEpochsForPartitions.get(tp).getOrElse {
    throw new IllegalStateException(
    s"Leader replied with partition $tp not requested in OffsetsForLeaderEpoch request")
    }
    //获取请求中携带的Follower副本的LeaderEpoch值
    val leaderEpochInRequest = partitionEpochRequest.currentLeaderEpoch.get
    //如果分区状态不为null，且请求携带的LeaderEpoch和当前分区状态中的LeaderEpoch一致，即发送请求的过程中没有发生leader副本的切换
    curPartitionState != null && leaderEpochInRequest == curPartitionState.currentLeaderEpoch
}

val ResultWithPartitions(fetchOffsets, partitionsWithError) = maybeTruncateToEpochEndOffsets(epochEndOffsets, latestEpochsForPartitions)

//处理出错的分区，即将分区对象从集合头部移除，然后放到集合的末尾
handlePartitionsWithErrors(partitionsWithError, "truncateToEpochEndOffsets")
//更新分区的状态
updateFetchOffsetAndMaybeMarkTruncationComplete(fetchOffsets)