深入理解Kafka服务端之日志对象的初始化及日志段的加载流程

class Log(@volatile var dir: File,//即主题分区的路径，每个主题的每个分区对应一个log对象
          @volatile var config: LogConfig,
          @volatile var logStartOffset: Long,//当前分区日志的起始偏移量
          @volatile var recoveryPoint: Long,
          scheduler: Scheduler,
          brokerTopicStats: BrokerTopicStats,
          val time: Time,
          val maxProducerIdExpirationMs: Int,
          val producerIdExpirationCheckIntervalMs: Int,
          val topicPartition: TopicPartition,
          val producerStateManager: ProducerStateManager,
          logDirFailureChannel: LogDirFailureChannel) extends Logging with KafkaMetricsGroup {
         ...
}

@volatile private var isMemoryMappedBufferClosed = false

@volatile private var nextOffsetMetadata: LogOffsetMetadata = _

@volatile private var replicaHighWatermark: Option[Long] = None

private val segments: ConcurrentNavigableMap[java.lang.Long, LogSegment] = new ConcurrentSkipListMap[java.lang.Long, LogSegment]

@volatile var leaderEpochCache: Option[LeaderEpochFileCache] = None

def recordVersion: RecordVersion = config.messageFormatVersion.recordVersion

locally {
    //记录起始时间
    val startMs = time.milliseconds
    //TODO 步骤一：创建分区日志路径对应的目录
    Files.createDirectories(dir.toPath)
    //TODO 步骤二：初始化LeaderEpochCache，生成检查点文件和创建LeaderEpochCache对象
    initializeLeaderEpochCache()
    //TODO 步骤三：加载所有日志段对象，返回下一条待写入消息的偏移量
    val nextOffset = loadSegments()
    //TODO 步骤四：更新nextOffsetMetadata 和 LogStartOffset
    nextOffsetMetadata = new LogOffsetMetadata(nextOffset, activeSegment.baseOffset, activeSegment.size)


    leaderEpochCache.foreach(_.truncateFromEnd(nextOffsetMetadata.messageOffset))


    logStartOffset = math.max(logStartOffset, segments.firstEntry.getValue.baseOffset)
    //TODO 步骤五：更新leaderEpochCache，清除无效数据
    leaderEpochCache.foreach(_.truncateFromStart(logStartOffset))
    
    if (!producerStateManager.isEmpty)
      throw new IllegalStateException("Producer state must be empty during log initialization")
    loadProducerState(logEndOffset, reloadFromCleanShutdown = hasCleanShutdownFile)


    info(s"Completed load of log with ${segments.size} segments, log start offset $logStartOffset and " +
      s"log end offset $logEndOffset in ${time.milliseconds() - startMs} ms")
  }

Files.createDirectories(dir.toPath)

initializeLeaderEpochCache()

private def initializeLeaderEpochCache(): Unit = lock synchronized {
  //创建leaderEpoch检查点文件对应的File对象。路径：分区路径/leader-epoch-checkpoint
  val leaderEpochFile = LeaderEpochCheckpointFile.newFile(dir)


  //创建 LeaderEpochFileCache的方法
  def newLeaderEpochFileCache(): LeaderEpochFileCache = {
    //创建检查点文件
    val checkpointFile = new LeaderEpochCheckpointFile(leaderEpochFile, logDirFailureChannel)
    //生成LeaderEpochFileCache对象
    new LeaderEpochFileCache(topicPartition, logEndOffset _, checkpointFile)
  }


  //如果是低于V2版本格式的日志，leaderEpochCache最终设置为None，因为0.11版本才引入epoch，该版本对应的日志格式为V2
  if (recordVersion.precedes(RecordVersion.V2)) {
    //如果leaderEpochFile对象已存在，返回一个LeaderEpochFileCache对象，否则返回None
    val currentCache = if (leaderEpochFile.exists())
      Some(newLeaderEpochFileCache())
    else
      None


    if (currentCache.exists(_.nonEmpty))
      warn(s"Deleting non-empty leader epoch cache due to incompatible message format $recordVersion")
    //如果leaderEpochFile对象对应的文件已存在，删除该文件
    Files.deleteIfExists(leaderEpochFile.toPath)


    leaderEpochCache = None
  //对于V2格式的日志，leaderEpochCache最终指向一个新创建的LeaderEpochFileCache对象
  } else {
    //生成LeaderEpochFileCache对象
    leaderEpochCache = Some(newLeaderEpochFileCache())
  }
}

//创建leaderEpoch检查点文件对应的File对象。路径：分区路径/leader-epoch-checkpoint
val leaderEpochFile = LeaderEpochCheckpointFile.newFile(dir)

if (recordVersion.precedes(RecordVersion.V2))

//生成LeaderEpochFileCache对象
leaderEpochCache = Some(newLeaderEpochFileCache())

val nextOffset = loadSegments()

private def loadSegments(): Long = {
    //步骤一：删除临时文件（包括.cleaned、.deleted 文件等），返回待恢复的文件集合，这些文件都是 xxx.log.swap 格式
    val swapFiles = removeTempFilesAndCollectSwapFiles()
    //步骤二：
    retryOnOffsetOverflow {
      //关闭所有的日志段对应的日志和索引文件
      logSegments.foreach(_.close())
      //清理segments集合中的日志段对象
      segments.clear()
      //重新加载日志段文件
      loadSegmentFiles()
    }
    //步骤三：处理有效的.swap文件，即将有效的swap文件对应的日志段添加到Log 的 segments 集合中
    completeSwapOperations(swapFiles)


    if (!dir.getAbsolutePath.endsWith(Log.DeleteDirSuffix)) {
      val nextOffset = retryOnOffsetOverflow {
        //步骤四：
        recoverLog()
      }
      //设置active 日志段 对应的索引文件大小为 10M
      activeSegment.resizeIndexes(config.maxIndexSize)
      //返回下一条待写入消息的偏移量
      nextOffset
      //如果目录以 -delete 结尾
    } else {
       if (logSegments.isEmpty) {
          addSegment(LogSegment.open(dir = dir,
            baseOffset = 0,
            config,
            time = time,
            fileAlreadyExists = false,
            initFileSize = this.initFileSize,
            preallocate = false))
       }
      0
    }
  }

val swapFiles = removeTempFilesAndCollectSwapFiles()

private def removeTempFilesAndCollectSwapFiles(): Set[File] = {
  // 在方法内部定义一个名为deleteIndicesIfExist的方法，用于删除日志文件对应的索引文件
  //该方法会在后面被调用
  def deleteIndicesIfExist(baseFile: File, suffix: String = ""): Unit = {
    info(s"Deleting index files with suffix $suffix for baseFile $baseFile")
    val offset = offsetFromFile(baseFile)
    Files.deleteIfExists(Log.offsetIndexFile(dir, offset, suffix).toPath)
    Files.deleteIfExists(Log.timeIndexFile(dir, offset, suffix).toPath)
    Files.deleteIfExists(Log.transactionIndexFile(dir, offset, suffix).toPath)
  }


  //定义几个保存不同类型文件的集合
  var swapFiles = Set[File]()//待恢复文件的集合
  var cleanFiles = Set[File]()//待删除文件的集合
  var minCleanedFileOffset = Long.MaxValue
  // 遍历分区日志路径下的所有文件
  for (file <- dir.listFiles if file.isFile) {
    // 如果不可读，直接抛出IOException
    if (!file.canRead)
      throw new IOException(s"Could not read file $file")
    val filename = file.getName
    // 如果以.deleted结尾
    if (filename.endsWith(DeletedFileSuffix)) {
      debug(s"Deleting stray temporary file ${file.getAbsolutePath}")
      //说明是上次Failure遗留下来的文件，直接删除
      Files.deleteIfExists(file.toPath)
      // 如果以.cleaned结尾，表示在执行日志压缩过程中宕机，文件中的数据状态不明确，无法正确恢复的文件。
    } else if (filename.endsWith(CleanedFileSuffix)) {
      //判断该文件名中的偏移量和minCleanedFileOffset 的大小，如果较小，则更新 minCleanedFileOffset
      minCleanedFileOffset = Math.min(offsetFromFileName(filename), minCleanedFileOffset)
      //将该文件加入待删除的文件集合
      cleanFiles += file
      // 如果以.swap结尾的文件，则说明日志压缩过程已完成，但是在执行交换过程中宕机
      // 1. 如果 swap 文件是 log 文件，则删除对应的 index 文件，稍后 swap 操作会重建索引
      // 2. 如果 swap 文件是 index 文件，则直接删除，后续加载 log 文件时会重建索引
    } else if (filename.endsWith(SwapFileSuffix)) {
      //去掉文件名中的.swap后缀
      val baseFile = new File(CoreUtils.replaceSuffix(file.getPath, SwapFileSuffix, ""))
      info(s"Found file ${file.getAbsolutePath} from interrupted swap operation.")
      // 如果该.swap文件原来是索引文件
      if (isIndexFile(baseFile)) {
        //删除原索引文件，之后会重建索引文件
        deleteIndicesIfExist(baseFile)
        //如果该.swap文件原来是日志文件
      } else if (isLogFile(baseFile)) {
        //删除原日志文件对应的索引文件，之后会重建索引文件
        deleteIndicesIfExist(baseFile)
        //加入待恢复的.swap文件集合中，这个集合中保存的都是 xxxx.log.swap 文件
        swapFiles += file
      }
    }
  }
  // 从待恢复swap集合中找出那些起始位移值大于minCleanedFileOffset值的文件，直接删掉这些无效的.swap文件
  val (invalidSwapFiles, validSwapFiles) = swapFiles.partition(file => offsetFromFile(file) >= minCleanedFileOffset)
  invalidSwapFiles.foreach { file =>
    debug(s"Deleting invalid swap file ${file.getAbsoluteFile} minCleanedFileOffset: $minCleanedFileOffset")
    val baseFile = new File(CoreUtils.replaceSuffix(file.getPath, SwapFileSuffix, ""))
    //删除日志对应的索引文件
    deleteIndicesIfExist(baseFile, SwapFileSuffix)
    //删除日志文件
    Files.deleteIfExists(file.toPath)
  }
  //清除所有待删除文件集合中的文件
  cleanFiles.foreach { file =>
    debug(s"Deleting stray .clean file ${file.getAbsolutePath}")
    Files.deleteIfExists(file.toPath)
  }
  //最后返回当前有效的.swap文件集合
  validSwapFiles
}

// 从待恢复swap集合中找出那些起始位移值大于minCleanedFileOffset值的文件，直接删掉这些无效的.swap文件
val (invalidSwapFiles, validSwapFiles) = swapFiles.partition(file => offsetFromFile(file) >= minCleanedFileOffset)
invalidSwapFiles.foreach { file =>
  debug(s"Deleting invalid swap file ${file.getAbsoluteFile} minCleanedFileOffset: $minCleanedFileOffset")
  val baseFile = new File(CoreUtils.replaceSuffix(file.getPath, SwapFileSuffix, ""))
  //删除日志对应的索引文件
  deleteIndicesIfExist(baseFile, SwapFileSuffix)
  //删除日志文件
  Files.deleteIfExists(file.toPath)
}

retryOnOffsetOverflow {
  //关闭所有的日志段对应的日志和索引文件
  logSegments.foreach(_.close())
  //清理segments集合中的日志段对象
  segments.clear()
  //重新加载日志段文件
  loadSegmentFiles()
}

private def loadSegmentFiles(): Unit = {
  //遍历分区目录下的文件
  for (file <- dir.listFiles.sortBy(_.getName) if file.isFile) {
    //如果是索引文件
    if (isIndexFile(file)) {
      //获取索引文件名中的起始偏移量
      val offset = offsetFromFile(file)
      //获取起始偏移量对应的日志文件，如果日志文件不存在，则删除该索引文件
      val logFile = Log.logFile(dir, offset)
      if (!logFile.exists) {
        warn(s"Found an orphaned index file ${file.getAbsolutePath}, with no corresponding log file.")
        Files.deleteIfExists(file.toPath)
      }
      //如果是日志文件
    } else if (isLogFile(file)) {
      //获取日志文件名中的起始偏移量
      val baseOffset = offsetFromFile(file)
      //标记日志对应的时间戳索引文件是否存在，如果存在为false
      val timeIndexFileNewlyCreated = !Log.timeIndexFile(dir, baseOffset).exists()
      //构建对应的LogSegment对象
      val segment = LogSegment.open(dir = dir,
        baseOffset = baseOffset,
        config,
        time = time,
        fileAlreadyExists = true)
      //如果偏移量索引文件存在，看是否需要重建对应的时间戳索引文件，并创建已中止事务索引文件
      //如果偏移量索引文件不存在，则抛异常
      try segment.sanityCheck(timeIndexFileNewlyCreated)
      catch {
        case _: NoSuchFileException =>
          error(s"Could not find offset index file corresponding to log file ${segment.log.file.getAbsolutePath}, " +
            "recovering segment and rebuilding index files...")
          recoverSegment(segment)
        case e: CorruptIndexException =>
          warn(s"Found a corrupted index file corresponding to log file ${segment.log.file.getAbsolutePath} due " +
            s"to ${e.getMessage}}, recovering segment and rebuilding index files...")
          recoverSegment(segment)
      }
      //将这个日志段对象添加到segments集合中
      addSegment(segment)
    }
  }
}

completeSwapOperations(swapFiles)

private def completeSwapOperations(swapFiles: Set[File]): Unit = {
  遍历文件
  for (swapFile <- swapFiles) {
    去掉.swap 后缀
    val logFile = new File(CoreUtils.replaceSuffix(swapFile.getPath, SwapFileSuffix, ""))
    获取起始偏移量
    val baseOffset = offsetFromFile(logFile)
    /创建LogSegment对象
    val swapSegment = LogSegment.open(swapFile.getParentFile,
      baseOffset = baseOffset,
      config,
      time = time,
      fileSuffix = SwapFileSuffix)
    info(s"Found log file ${swapFile.getPath} from interrupted swap operation, repairing.")
    //恢复日志段，重建索引文件，并校验日志中消息的合法性
    recoverSegment(swapSegment)


    // 查找 swapSegment 获取 [baseOffset, nextOffset] 区间对应的日志压缩前的 LogSegment 集合，
    // 区间中的 LogSegment 数据都压缩到了 swapSegment 中
    val oldSegments = logSegments(swapSegment.baseOffset, swapSegment.readNextOffset).filter { segment =>
      segment.readNextOffset > swapSegment.baseOffset
    }
    //将压缩前的日志段对应的LogSegment对象从segments集合中移除，并删除对应的日志和索引文件
    // 然后将新创建的swapSegment日志段添加到Log对象的segments中，最后移除文件的.swap 后缀
    replaceSegments(Seq(swapSegment), oldSegments.toSeq, isRecoveredSwapFile = true)
  }
}

if (!dir.getAbsolutePath.endsWith(Log.DeleteDirSuffix)) {
  val nextOffset = retryOnOffsetOverflow {
    //处理 broker 节点异常关闭导致的数据异常，需要验证 [recoveryPoint, Long.MaxValue] 中的所有消息，并移除验证失败的消息
    recoverLog()
  }
  //设置active 日志段 对应的索引文件大小为 10M
  activeSegment.resizeIndexes(config.maxIndexSize)
  //返回下一条待写入消息的偏移量
  nextOffset
  //如果目录以 -delete 结尾
} else {
   if (logSegments.isEmpty) {
      addSegment(LogSegment.open(dir = dir,
        baseOffset = 0,
        config,
        time = time,
        fileAlreadyExists = false,
        initFileSize = this.initFileSize,
        preallocate = false))
   }
  0
}

private def recoverLog(): Long = {
  //判断是否存在.kafka_cleanshutdown 文件，如果存在则说明broker异常关闭
  //需要对recoveryPoint恢复点之后的数据进行校验，如果不完整则丢弃
  if (!hasCleanShutdownFile) {
    // okay we need to actually recover this log
    //获取包含恢复点之后消息的 日志段对象集合
    val unflushed = logSegments(this.recoveryPoint, Long.MaxValue).iterator


    while (unflushed.hasNext) {
      val segment = unflushed.next
      info(s"Recovering unflushed segment ${segment.baseOffset}")
      val truncatedBytes =
        try {
          //恢复日志段
          recoverSegment(segment, leaderEpochCache)
        } catch {
          case _: InvalidOffsetException =>
            val startOffset = segment.baseOffset
            warn("Found invalid offset during recovery. Deleting the corrupt segment and " +
              s"creating an empty one with starting offset $startOffset")
            segment.truncateTo(startOffset)
        }
      //如果有无效的消息，直接删除所有的未刷写日志段
      if (truncatedBytes > 0) {
        // we had an invalid message, delete all remaining log
        warn(s"Corruption found in segment ${segment.baseOffset}, truncating to offset ${segment.readNextOffset}")
        //删除日志段
        unflushed.foreach(deleteSegment)
      }
    }
  }
  //执行完之后，如果日志段集合不为空
  if (logSegments.nonEmpty) {
    //获取active 日志段的LEO
    val logEndOffset = activeSegment.readNextOffset
    //如果LEO 比 日志的 LogStartOffset 还小，则删除所有日志段
    if (logEndOffset < logStartOffset) {
      warn(s"Deleting all segments because logEndOffset ($logEndOffset) is smaller than logStartOffset ($logStartOffset). " +
        "This could happen if segment files were deleted from the file system.")
      logSegments.foreach(deleteSegment)
    }
  }
  //执行完上面的逻辑，如果日志段集合已经为空
  if (logSegments.isEmpty) {
    //至少要有一个active 日志段，上面全部删除了，所以要创建一个新的日志段，起始位移就是日志的LogStartOffset，然后放入segments集合
    addSegment(LogSegment.open(dir = dir,
      baseOffset = logStartOffset,
      config,
      time = time,
      fileAlreadyExists = false,
      initFileSize = this.initFileSize,
      preallocate = config.preallocate))
  }
  //更新恢复点信息，返回恢复点
  recoveryPoint = activeSegment.readNextOffset
  recoveryPoint
}

nextOffsetMetadata = new LogOffsetMetadata(nextOffset, activeSegment.baseOffset, activeSegment.size)


leaderEpochCache.foreach(_.truncateFromEnd(nextOffsetMetadata.messageOffset))


logStartOffset = math.max(logStartOffset, segments.firstEntry.getValue.baseOffset)

leaderEpochCache.foreach(_.truncateFromStart(logStartOffset))