class Partition(val topicPartition: TopicPartition, // topic分区标识
                val replicaLagTimeMaxMs: Long, // 副本最长延迟时间
                interBrokerProtocolVersion: ApiVersion,
                localBrokerId: Int, // 本地broker id
                time: Time,
                stateStore: PartitionStateStore, // 存储在ZooKeeper的分区元数据存储
                delayedOperations: DelayedOperations,
                metadataCache: MetadataCache,
                logManager: LogManager // 日志处理器
               ) extends Logging with KafkaMetricsGroup {
  // ...
  private val remoteReplicasMap = new Pool[Int, Replica] // 远程副本集合，维护ID-副本的映射
  private val leaderIsrUpdateLock = new ReentrantReadWriteLock
  private var zkVersion: Int = LeaderAndIsr.initialZKVersion
  @volatile private var leaderEpoch: Int = LeaderAndIsr.initialLeaderEpoch - 1
  @volatile private var leaderEpochStartOffsetOpt: Option[Long] = None
  @volatile var leaderReplicaIdOpt: Option[Int] = None // 主副本的副本ID
  @volatile var inSyncReplicaIds = Set.empty[Int] // 正在同步的副本集（ISR），以ID形式保存
  @volatile var assignmentState: AssignmentState = SimpleAssignmentState(Seq.empty) // 分配给该分区的所有副本ID
  @volatile var log: Option[Log] = None  // 该分区的日志
  @volatile var futureLog: Option[Log] = None
  private var controllerEpoch: Int = KafkaController.InitialControllerEpoch
  this.logIdent = s"[Partition $topicPartition broker=$localBrokerId] "
  private val tags = Map("topic" -> topic, "partition" -> partitionId.toString)
  // ...
}

class Replica(val brokerId: Int, // broker id
              val topicPartition: TopicPartition // topic分区标识
             ) extends Logging {
  // 日志尾部的offset值（本地副本即日志尾部，远程副本只会在从副本拉取后更新）
  // 包含日志offset和文件偏移的位置
  @volatile private[this] var _logEndOffsetMetadata = LogOffsetMetadata.UnknownOffsetMetadata
  // 日志起始offset值（本地副本即日志头部，远程副本只会在从副本拉取后更新）
  @volatile private[this] var _logStartOffset = Log.UnknownOffset
  // 最近一次从副本请求拉取更新的日志尾offset值
  @volatile private[this] var lastFetchLeaderLogEndOffset = 0L
  // 上一次主副本收到从副本拉取请求的时间
  @volatile private[this] var lastFetchTimeMs = 0L
  // 从副本需要追上主副本所需要点最大时间
  @volatile private[this] var _lastCaughtUpTimeMs = 0L
  // 从副本上一次看到主副本的高水位值
  @volatile private[this] var _lastSentHighWatermark = 0L
}

class Log(@volatile var dir: File, // 日志目录
          @volatile var config: LogConfig,
          @volatile var logStartOffset: Long, // 日志开头的offset
          @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 {
  private val lock = new Object
  @volatile private var isMemoryMappedBufferClosed = false
  // 上一次刷盘点时间
  private val lastFlushedTime = new AtomicLong(time.milliseconds)
  // 下一个消息对应的offset
  @volatile private var nextOffsetMetadata: LogOffsetMetadata = _
  // 第一个不稳定（未完成事务的）偏移量元数据
  @volatile private var firstUnstableOffsetMetadata: Option[LogOffsetMetadata] = None
  // 高水位元数据
  @volatile private var highWatermarkMetadata: LogOffsetMetadata = LogOffsetMetadata(logStartOffset)
  // 日志包括多个日志分段
  // 它用一个有序Map（跳表）保存，键是起始偏移，最后一个就是最新、活跃的日志分段
  private val segments: ConcurrentNavigableMap[java.lang.Long, LogSegment] = new ConcurrentSkipListMap[java.lang.Long, LogSegment]
  // 主副本的Epoch
  @volatile var leaderEpochCache: Option[LeaderEpochFileCache] = None 
}

class LogSegment private[log] (val log: FileRecords, // 数据文件
                               val lazyOffsetIndex: LazyIndex[OffsetIndex], // 索引文件
                               val lazyTimeIndex: LazyIndex[TimeIndex],
                               val txnIndex: TransactionIndex,
                               val baseOffset: Long, // 该分段最早的offset
                               val indexIntervalBytes: Int,
                               val rollJitterMs: Long,
                               val time: Time) extends Logging {
  // ...
}

public class MemoryRecords extends AbstractRecords {
    private final ByteBuffer buffer; // 消息数据
    private final Iterable<MutableRecordBatch> batches = this::batchIterator; // 消息迭代器
    private int validBytes = -1; //
  
    // ...
  
    @Override
    public AbstractIterator<MutableRecordBatch> batchIterator() {
        return new RecordBatchIterator<>(new ByteBufferLogInputStream(buffer.duplicate(), Integer.MAX_VALUE));
    }
  
    // ...
}

private def append(records: MemoryRecords, // 一批二进制的RecordBatch
                   origin: AppendOrigin,
                   interBrokerProtocolVersion: ApiVersion,
                   assignOffsets: Boolean,
                   leaderEpoch: Int): LogAppendInfo = {
  maybeHandleIOException(s"Error while appending records to $topicPartition in dir ${dir.getParent}") {
    // ... 
    var validRecords = trimInvalidBytes(records, appendInfo)
    lock synchronized {
      checkIfMemoryMappedBufferClosed()
      if (assignOffsets) {
        // 获取下一批写入应有的起始offset
        val offset = new LongRef(nextOffsetMetadata.messageOffset)
        appendInfo.firstOffset = Some(offset.value)
        val now = time.milliseconds
        // 修正设置每一个RecordBatch的start offset
        val validateAndOffsetAssignResult = try {
          LogValidator.validateMessagesAndAssignOffsets(validRecords,
            topicPartition,
            offset,
            time,
            now,
            appendInfo.sourceCodec,
            appendInfo.targetCodec,
            config.compact,
            config.messageFormatVersion.recordVersion.value,
            config.messageTimestampType,
            config.messageTimestampDifferenceMaxMs,
            leaderEpoch,
            origin,
            interBrokerProtocolVersion,
            brokerTopicStats)
        } catch {
          case e: IOException =>
            throw new KafkaException(s"Error validating messages while appending to log $name", e)
        }
        // ...
      }
    }
  }
}

private def assignOffsetsNonCompressed(records: MemoryRecords, // 多个RecordBatch
                                       topicPartition: TopicPartition,
                                       // 初始为分区全局的下一个Record的绝对偏移量，即nextOffsetMetadata
                                       offsetCounter: LongRef,
                                       now: Long,
                                       compactedTopic: Boolean,
                                       timestampType: TimestampType,
                                       timestampDiffMaxMs: Long,
                                       partitionLeaderEpoch: Int,
                                       origin: AppendOrigin,
                                       magic: Byte, // 等于2
                                       brokerTopicStats: BrokerTopicStats): ValidationAndOffsetAssignResult = {
  var maxTimestamp = RecordBatch.NO_TIMESTAMP
  var offsetOfMaxTimestamp = -1L
  val initialOffset = offsetCounter.value
  val firstBatch = getFirstBatchAndMaybeValidateNoMoreBatches(records, NoCompressionCodec)
  // 遍历RecordBatch
  for (batch <- records.batches.asScala) {
    // 验证RecordBatch
    validateBatch(topicPartition, firstBatch, batch, origin, magic, brokerTopicStats)
    var maxBatchTimestamp = RecordBatch.NO_TIMESTAMP
    var offsetOfMaxBatchTimestamp = -1L
    val recordErrors = new ArrayBuffer[ApiRecordError](0)
    // 遍历Record
    for ((record, batchIndex) <- batch.asScala.view.zipWithIndex) {
      // 验证Record
      validateRecord(batch, topicPartition, record, batchIndex, now, timestampType,
        timestampDiffMaxMs, compactedTopic, brokerTopicStats).foreach(recordError => recordErrors += recordError)
      // 增加offset值（每次+1），代表最新的Record偏移量
      val offset = offsetCounter.getAndIncrement()
      if (batch.magic > RecordBatch.MAGIC_VALUE_V0 && record.timestamp > maxBatchTimestamp) {
        maxBatchTimestamp = record.timestamp
        offsetOfMaxBatchTimestamp = offset
      }
    }
    processRecordErrors(recordErrors)
    if (batch.magic > RecordBatch.MAGIC_VALUE_V0 && maxBatchTimestamp > maxTimestamp) {
      maxTimestamp = maxBatchTimestamp
      offsetOfMaxTimestamp = offsetOfMaxBatchTimestamp
    }
    // 这里设置了RecordBatch的start offset
    batch.setLastOffset(offsetCounter.value - 1)
    // 设置其他字段...
    // ...
  }
  // ...
  ValidationAndOffsetAssignResult(...)
}

public void setLastOffset(long offset) {
    // 最新的绝对offset - 最新的delta offset = 第一个的offset
    buffer.putLong(BASE_OFFSET_OFFSET, offset - lastOffsetDelta());
}

def shouldRoll(rollParams: RollParams): Boolean = {
  val reachedRollMs = timeWaitedForRoll(rollParams.now, rollParams.maxTimestampInMessages) > rollParams.maxSegmentMs - rollJitterMs
  // 创建分段的条件：
  // 1. 分段大小过大（log.segment.bytes）
  size > rollParams.maxSegmentBytes - rollParams.messagesSize ||
    (size > 0 && reachedRollMs) || // 2. 分段维持了一定时间（log.roll.ms）
    offsetIndex.isFull || timeIndex.isFull || // 3. 索引文件满（log.index.size.max.bytes）
    !canConvertToRelativeOffset(rollParams.maxOffsetInMessages) // 4. 追加消息的偏移量与当前分段的基准偏移量大于Integer.MAX_VALUE
}

def roll(expectedNextOffset: Option[Long] = None): LogSegment = {
  maybeHandleIOException(s"Error while rolling log segment for $topicPartition in dir ${dir.getParent}") {
    val start = time.hiResClockMs()
    lock synchronized {
      checkIfMemoryMappedBufferClosed()
      // 设置最新偏移量为新分段的起始偏移量
      val newOffset = math.max(expectedNextOffset.getOrElse(0L), logEndOffset)
      // 创建新分段数据文件，以起始偏移量作为文件名
      val logFile = Log.logFile(dir, newOffset)
      // ...
      } else {
      	// 创建偏移量索引文件
        val offsetIdxFile = offsetIndexFile(dir, newOffset)
        // 创建时间索引文件
        val timeIdxFile = timeIndexFile(dir, newOffset)
        // 创建事务索引文件
        val txnIdxFile = transactionIndexFile(dir, newOffset)
				// ...
        // 触发让分段不活跃的回调
        Option(segments.lastEntry).foreach(_.getValue.onBecomeInactiveSegment())
      }
      // ...
      // 创建新分段
      val segment = LogSegment.open(dir,
        baseOffset = newOffset,
        config,
        time = time,
        fileAlreadyExists = false,
        initFileSize = initFileSize,
        preallocate = config.preallocate)
      // 添加新分段到日志
      addSegment(segment)
      // 更新日志尾的偏移量
      updateLogEndOffset(nextOffsetMetadata.messageOffset)
      // 启动后台线程，将旧分段数据刷盘
      scheduler.schedule("flush-log", () => flush(newOffset), delay = 0L)
      segment
    }
  }
}

def append(largestOffset: Long,
           largestTimestamp: Long,
           shallowOffsetOfMaxTimestamp: Long,
           records: MemoryRecords): Unit = {
  if (records.sizeInBytes > 0) {
    // 读取文件物理位置
    val physicalPosition = log.sizeInBytes()
    if (physicalPosition == 0)
      rollingBasedTimestamp = Some(largestTimestamp)
    ensureOffsetInRange(largestOffset)
    // 追加数据文件，返回写入字节数
    val appendedBytes = log.append(records)
    if (largestTimestamp > maxTimestampSoFar) {
      maxTimestampSoFar = largestTimestamp
      offsetOfMaxTimestampSoFar = shallowOffsetOfMaxTimestamp
    }
    // 若距离上次写入索引项时，已经累积写入超过indexIntervalBytes的数据项，则写入索引项
    if (bytesSinceLastIndexEntry > indexIntervalBytes) {
      offsetIndex.append(largestOffset, physicalPosition)
      timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar)
      bytesSinceLastIndexEntry = 0 // 清零
    }
    bytesSinceLastIndexEntry += records.sizeInBytes // 累加数据项大小
  }
}

def read(startOffset: Long, // 给定的起始偏移量
         maxLength: Int, // 最大拉取的数据大小
         isolation: FetchIsolation, // 拉取隔离配置: LogEnd, HighWatermark, TxnCommitted
         minOneMessage: Boolean // 若为true，若第一个消息大于maxLength，则会返回该消息
        ): FetchDataInfo = {
  maybeHandleIOException(s"Exception while reading from $topicPartition in dir ${dir.getParent}") {
    val includeAbortedTxns = isolation == FetchTxnCommitted
    // 获取当前日志尾偏移量
    val endOffsetMetadata = nextOffsetMetadata 
    val endOffset = nextOffsetMetadata.messageOffset
    if (startOffset == endOffset)
    	// 若起始偏移=尾偏移，则没消息可拉取，直接返回
      return emptyFetchDataInfo(endOffsetMetadata, includeAbortedTxns)
    // 二分法，调用floor，从跳表中获取对应的日志分段
    var segmentEntry = segments.floorEntry(startOffset)
    // ...
    // 根据isolation，确定拉取消息偏移量的上限，即：
    // a) FetchLogEnd: 上限为日志尾偏移量
    // b) FetchHighWatermark: 上限为最高水位(一般小于日志尾偏移量)，该水位前的数据已经复制到了给定数量多从副本
    // c) FetchTxnCommitted: 上限为最后一个事物提及的偏移量
    val maxOffsetMetadata = isolation match {
      case FetchLogEnd => nextOffsetMetadata
      case FetchHighWatermark => fetchHighWatermarkMetadata
      case FetchTxnCommitted => fetchLastStableOffsetMetadata
    }
    
    // 若给定的起始偏移量超过了最大偏移量，则没有消息读到，返回空
    if (startOffset > maxOffsetMetadata.messageOffset) {
      val startOffsetMetadata = convertToOffsetMetadataOrThrow(startOffset)
      return emptyFetchDataInfo(startOffsetMetadata, includeAbortedTxns)
    }
    
    while (segmentEntry != null) {
      val segment = segmentEntry.getValue
      val maxPosition = {
        if (maxOffsetMetadata.segmentBaseOffset == segment.baseOffset) {
          maxOffsetMetadata.relativePositionInSegment
        } else {
          segment.size
        }
      }
      // 从对应分段里读消息
      val fetchInfo = segment.read(startOffset, maxLength, maxPosition, minOneMessage)
      if (fetchInfo == null) {
        // 若没读到数据，则从下一个分段里再读一次
        segmentEntry = segments.higherEntry(segmentEntry.getKey)
      } else {
        // 否则，返回读取到的消息
        return if (includeAbortedTxns)
          addAbortedTransactions(startOffset, segmentEntry, fetchInfo)
        else
          fetchInfo
      }
    }
    FetchDataInfo(nextOffsetMetadata, MemoryRecords.EMPTY)
  }
}

// In LogSegment
@threadsafe
private[log] def translateOffset(offset: Long, startingFilePosition: Int = 0): LogOffsetPosition = {
  // 1. 利用mmap内存文件映射，使用二分法，返回目标消息的绝对偏移量和物理位置
  val mapping = offsetIndex.lookup(offset)
  // 2. 然后利用查询到的物理位置，返回其所在的RecordBatch的最后一项的绝对偏移、物理位置和大小
  log.searchForOffsetWithSize(offset, max(mapping.position, startingFilePosition))
}

// In OffsetIndex
// 查询索引文件，找到目标offset在数据文件中的物理位置
def lookup(targetOffset: Long): OffsetPosition = {
  maybeLock(lock) {
    val idx = mmap.duplicate // 首先将mmap duplicate一份（逻辑复制）
    val slot = largestLowerBoundSlotFor(idx, targetOffset, IndexSearchType.KEY) // 二分法找到索引项在索引文件中的下标
    if(slot == -1)
      OffsetPosition(baseOffset, 0)
    else
      // 转换成 (绝对偏移量, 物理位置)，并返回
      parseEntry(idx, slot)
  }
}

override protected def parseEntry(buffer: ByteBuffer, n: Int): OffsetPosition = {
  // 绝对偏移量 = 基准偏移量 + 相对偏移量（即mmap.getInt(slot * indexEntrySize)）
  // 物理位置 = mmap.getInt(slot * indexEntrySize + 4)
  OffsetPosition(baseOffset + relativeOffset(buffer, n), physical(buffer, n))
}

// In FileRecords
public LogOffsetPosition searchForOffsetWithSize(long targetOffset, int startingPosition) {
    // 从给定的起始位置开始遍历RecordBatch
    for (FileChannelRecordBatch batch : batchesFrom(startingPosition)) {
        // 返回第一个大于等于目标偏移的RecordBatch信息，即包含该消息的RecordBatch
        // 包括：RecordBatch最后一个消息的绝对偏移、RecordBatch的物理位置、RecordBatch的大小
        long offset = batch.lastOffset();
        if (offset >= targetOffset)
            return new LogOffsetPosition(offset, batch.position(), batch.sizeInBytes());
    }
    return null;
}

public FileRecords slice(int position, int size) throws IOException {
    // ...
    int end = this.start + position + size; // 设置文件末尾物理位置：原物理位置+新物理位置+长度
    if (end < 0 || end >= start + sizeInBytes()) // 处理整数溢出
        end = start + sizeInBytes();
    // 返回一个FileRecords逻辑视图，起始位置设为：原物理位置+新物理位置，末尾位置如上所述
    return new FileRecords(file, channel, this.start + position, end, true); 
}

// In LogSegment
def read(startOffset: Long,
         maxSize: Int,
         maxPosition: Long = size,
         minOneMessage: Boolean = false): FetchDataInfo = {
  // 1. 查找索引，得到记录所在的RecordBatch最后一项的绝对偏移量、RecordBatch的物理位置和大小
  val startOffsetAndSize = translateOffset(startOffset)
  if (startOffsetAndSize == null)
    return null
  val startPosition = startOffsetAndSize.position
  // 2. 设置偏移量元数据，包含: 所在RecordBatch的最后一项的绝对偏移量，当前日志分段的基准偏移量，所在RecordBatch的起始物理位置
  val offsetMetadata = LogOffsetMetadata(startOffset, this.baseOffset, startPosition)
  val adjustedMaxSize =
    if (minOneMessage) math.max(maxSize, startOffsetAndSize.size)
    else maxSize
  if (adjustedMaxSize == 0)
    return FetchDataInfo(offsetMetadata, MemoryRecords.EMPTY)
  val fetchSize: Int = min((maxPosition - startPosition).toInt, adjustedMaxSize)
  // 3. 调用slice，返回所需拉取数据的文件消息集视图，随着偏移量元数据，返回给上层
  FetchDataInfo(offsetMetadata, log.slice(startPosition, fetchSize),
    firstEntryIncomplete = adjustedMaxSize < startOffsetAndSize.size)
}

private val currentLogs = new Pool[TopicPartition, Log]() // 日志实例

def getOrCreateLog(topicPartition: TopicPartition, config: LogConfig, isNew: Boolean = false, isFuture: Boolean = false): Log = {
  logCreationOrDeletionLock synchronized {
    // 获取日志，没有则创建
    getLog(topicPartition, isFuture).getOrElse {
      // ...
      // 选择一个目录存储日志
      val logDirs: List[File] = {
        val preferredLogDir = preferredLogDirs.get(topicPartition)
        if (preferredLogDir != null)
          List(new File(preferredLogDir))
        else
          nextLogDirs()
      }
      // 根据topic和分区设置目录名
      val logDirName = {
        if (isFuture)
          Log.logFutureDirName(topicPartition)
        else
          Log.logDirName(topicPartition)
      }
		  // 创建目录
      val logDir = logDirs
        .toStream // to prevent actually mapping the whole list, lazy map
        .map(createLogDirectory(_, logDirName))
        .find(_.isSuccess)
        .getOrElse(Failure(new KafkaStorageException("No log directories available. Tried " + logDirs.map(_.getAbsolutePath).mkString(", "))))
        .get // If Failure, will throw
      // 创建日志实例
      val log = Log(
        dir = logDir,
        config = config,
        logStartOffset = 0L,
        recoveryPoint = 0L,
        maxProducerIdExpirationMs = maxPidExpirationMs,
        producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs,
        scheduler = scheduler,
        time = time,
        brokerTopicStats = brokerTopicStats,
        logDirFailureChannel = logDirFailureChannel)

      if (isFuture)
        futureLogs.put(topicPartition, log)
      else
        // 将日志加入到currentLogs中管理
        currentLogs.put(topicPartition, log)
      // ...
      preferredLogDirs.remove(topicPartition)
      log // 返回日志实例
    }
  }
}

private def loadLogs(): Unit = {
  val startMs = time.milliseconds
  val threadPools = ArrayBuffer.empty[ExecutorService]
  val offlineDirs = mutable.Set.empty[(String, IOException)]
  val jobs = mutable.Map.empty[File, Seq[Future[_]]]

  for (dir <- liveLogDirs) {
    try {
      // 给每个目录创建一个线程池
      val pool = Executors.newFixedThreadPool(numRecoveryThreadsPerDataDir)
      threadPools.append(pool)
      // ...
      // 读取检查点内容
      var recoveryPoints = Map[TopicPartition, Long]()
      try {
        recoveryPoints = this.recoveryPointCheckpoints(dir).read
      } catch {
        case e: Exception =>
          warn(s"Error occurred while reading recovery-point-offset-checkpoint file of directory $dir", e)
          warn("Resetting the recovery checkpoint to 0")
      }
      // 读取检查点中每个分区的起始偏移量
      var logStartOffsets = Map[TopicPartition, Long]()
      try {
        logStartOffsets = this.logStartOffsetCheckpoints(dir).read
      } catch {
        case e: Exception =>
          warn(s"Error occurred while reading log-start-offset-checkpoint file of directory $dir", e)
      }
      // 设置加载日志的目录
      val jobsForDir = for {
        dirContent <- Option(dir.listFiles).toList
        logDir <- dirContent if logDir.isDirectory
      } yield {
        // 创建加载日志的任务
        val runnable: Runnable = () => {
          try {
            loadLog(logDir, recoveryPoints, logStartOffsets) // 加载日志
          } catch {
            case e: IOException =>
              offlineDirs.add((dir.getAbsolutePath, e))
              error(s"Error while loading log dir ${dir.getAbsolutePath}", e)
          }
        }
        runnable
      }
      // 提交任务
      jobs(cleanShutdownFile) = jobsForDir.map(pool.submit)
    } catch {
      case e: IOException =>
        offlineDirs.add((dir.getAbsolutePath, e))
        error(s"Error while loading log dir ${dir.getAbsolutePath}", e)
    }
  }
  // ...
}

def checkpointLogRecoveryOffsets(): Unit = {
  // 遍历所有的log.dirs目录
  logsByDir.foreach { case (dir, partitionToLogMap) =>
    liveLogDirs.find(_.getAbsolutePath.equals(dir)).foreach { f =>
      // 在该目录下，刷新对应topic分区的检查点
      // partitionToLogMap.values是目录下所有的分区日志Log实例
      checkpointRecoveryOffsetsAndCleanSnapshot(f, partitionToLogMap.values.toSeq)
    }
  }
}

private[log] def checkpointRecoveryOffsetsAndCleanSnapshot(dir: File, logsToCleanSnapshot: Seq[Log]): Unit = {
  try {
    // 写入检查点文件
    checkpointLogRecoveryOffsetsInDir(dir)
    logsToCleanSnapshot.foreach(_.deleteSnapshotsAfterRecoveryPointCheckpoint())
  } catch {
    // .. 
  }
}

private def checkpointLogRecoveryOffsetsInDir(dir: File): Unit = {
  // 遍历每个分区，刷新检查点
  // 写入的检查点即日志最新刷盘的偏移量
  for {
    partitionToLog <- logsByDir.get(dir.getAbsolutePath)
    checkpoint <- recoveryPointCheckpoints.get(dir)
  } {
    checkpoint.write(partitionToLog.map { case (tp, log) => tp -> log.recoveryPoint })
  }
}

private def flushDirtyLogs(): Unit = {
  for ((topicPartition, log) <- currentLogs.toList ++ futureLogs.toList) {
    // 遍历所有日志
    try {
      // 若距离上次刷盘超过log.flush.interval.ms，则刷盘
      val timeSinceLastFlush = time.milliseconds - log.lastFlushTime
      if(timeSinceLastFlush >= log.config.flushMs)
        log.flush // 刷盘
    } catch {
      // ...
    }
  }
}

def flush(offset: Long): Unit = {
  maybeHandleIOException(s"Error while flushing log for $topicPartition in dir ${dir.getParent} with offset $offset") {
    if (offset <= this.recoveryPoint)
      return
    // 遍历需要刷盘的分段，先刷盘
    // 分段可能有多个（当上一个检查点不在最新的分段上时）
    for (segment <- logSegments(this.recoveryPoint, offset))
      segment.flush()

    lock synchronized {
      checkIfMemoryMappedBufferClosed()
      if (offset > this.recoveryPoint) {
        // 刷盘后，设置新检查点
        this.recoveryPoint = offset
        lastFlushedTime.set(time.milliseconds)
      }
    }
  }
}

def cleanupLogs(): Unit = {
  var total = 0
  val startMs = time.milliseconds
  // 得到没有压缩过的日志实例，作为可清理的对象
  val deletableLogs = {
    if (cleaner != null) {
      // 首先暂停日志清理
      // 然后返回没有压缩过，且没有正在清理的日志实例
      // 返回的日志实例，在cleaner中会被暂停清理
      cleaner.pauseCleaningForNonCompactedPartitions()
    } else {
      currentLogs.filter {
        case (_, log) => !log.config.compact
      }
    }
  }

  try {
    deletableLogs.foreach {
      // 遍历可清理的日志对象
      case (topicPartition, log) =>
        // 删除日志的旧分段
        total += log.deleteOldSegments()
        val futureLog = futureLogs.get(topicPartition)
        if (futureLog != null) {
          total += futureLog.deleteOldSegments()
        }
    }
  } finally {
    // 重新恢复日志的清理
    if (cleaner != null) {
      cleaner.resumeCleaning(deletableLogs.map(_._1))
    }
  }
}

def deleteOldSegments(): Int = {
  if (config.delete) {
    // 若配置策略为“删除”
    deleteRetentionMsBreachedSegments() // 1. 删除保留时间过长的日志分段
    + deleteRetentionSizeBreachedSegments() // 2. 删除保留大小过大的日志分段，日志整个大小
                                            // 删除的条件为: size - retentionSize - segment.size >= 0
    + deleteLogStartOffsetBreachedSegments() // 3. 删除分段末尾偏移量小于该分区日志（含多个分段）起始偏移量的日志分段
  } else {
    // 若配置策略为“压缩”
    // 只能删除比日志起始偏移量小的日志分段，后面的分段将会被压缩
    deleteLogStartOffsetBreachedSegments()
  }
}

private[log] class LogCleanerManager(val logDirs: Seq[File], // 所有日志目录
                                     val logs: Pool[TopicPartition, Log], // 所有日志
                                     val logDirFailureChannel: LogDirFailureChannel) extends Logging with KafkaMetricsGroup {
  import LogCleanerManager._
  protected override def loggerName = classOf[LogCleaner].getName
  // 清理点检查点文件名
  private[log] val offsetCheckpointFile = "cleaner-offset-checkpoint"
  // 所有日志的检查点文件
  @volatile private var checkpoints = logDirs.map(dir =>
    (dir, new OffsetCheckpointFile(new File(dir, offsetCheckpointFile), logDirFailureChannel))).toMap
  // ...
}

def grabFilthiestCompactedLog(time: Time, preCleanStats: PreCleanStats = new PreCleanStats()): Option[LogToClean] = {
  inLock(lock) {
    // ...
    val lastClean = allCleanerCheckpoints // 读取记录了清理点检查点文件
    // 过滤日志，过滤后得到的日志需要：
    // 1. 配置“压缩”清理策略
    // 2. 日志没有正在压缩中
    // 3. 需要压缩的日志旧分段非空
    val dirtyLogs = logs.filter {
      case (_, log) => log.config.compact
    }.filterNot {
      case (topicPartition, log) =>
        inProgress.contains(topicPartition) || isUncleanablePartition(log, topicPartition)
    }.map {
      // 为每个日志，创建LogToClean对象
      case (topicPartition, log) => 
        try {
          val lastCleanOffset = lastClean.get(topicPartition)
          val (firstDirtyOffset, firstUncleanableDirtyOffset) = cleanableOffsets(log, lastCleanOffset, now)
          val compactionDelayMs = maxCompactionDelay(log, firstDirtyOffset, now)
          preCleanStats.updateMaxCompactionDelay(compactionDelayMs)
          LogToClean(topicPartition, log, firstDirtyOffset, firstUncleanableDirtyOffset, compactionDelayMs > 0)
        } catch {
          case e: Throwable => throw new LogCleaningException(log,
            s"Failed to calculate log cleaning stats for partition $topicPartition", e)
        }
    }.filter(ltc => ltc.totalBytes > 0)
    // 计算最大比例：日志头部大小(dirtyBytes)/日志大小
    this.dirtiestLogCleanableRatio = if (dirtyLogs.nonEmpty) dirtyLogs.max.cleanableRatio else 0
    // 选择上面比例最大的日志，并且必须要满足最小比例阈值
    // 最后得到0或1个日志
    val cleanableLogs = dirtyLogs.filter { ltc =>
      (ltc.needCompactionNow && ltc.cleanableBytes > 0) || ltc.cleanableRatio > ltc.log.config.minCleanableRatio
    }
    // 返回
    if(cleanableLogs.isEmpty) {
      None
    } else {
      preCleanStats.recordCleanablePartitions(cleanableLogs.size)
      val filthiest = cleanableLogs.max
      inProgress.put(filthiest.topicPartition, LogCleaningInProgress) // 设置日志正在压缩
      Some(filthiest)
    }
  }
}

private case class LogToClean(topicPartition: TopicPartition,
                              log: Log,
                              firstDirtyOffset: Long, // 日志头部第一个偏移量
                              uncleanableOffset: Long, // 第一个不可清理的第一个偏移量
                              needCompactionNow: Boolean = false) extends Ordered[LogToClean] {
  // 日志尾部大小
  val cleanBytes = log.logSegments(-1, firstDirtyOffset).map(_.size.toLong).sum
  // 第一个不可清理的偏移量（它由活动分段确定），日志头部大小
  val (firstUncleanableOffset, cleanableBytes) = LogCleanerManager.calculateCleanableBytes(log, firstDirtyOffset, uncleanableOffset)
  // 待清理的日志总大小
  val totalBytes = cleanBytes + cleanableBytes
  // 比例计算
  val cleanableRatio = cleanableBytes / totalBytes.toDouble
  override def compare(that: LogToClean): Int = math.signum(this.cleanableRatio - that.cleanableRatio).toInt
}

private[log] class CleanerThread(threadId: Int)
  extends ShutdownableThread(name = s"kafka-log-cleaner-thread-$threadId", isInterruptible = false) {

  protected override def loggerName = classOf[LogCleaner].getName

  if (config.dedupeBufferSize / config.numThreads > Int.MaxValue)
    warn("Cannot use more than 2G of cleaner buffer space per cleaner thread, ignoring excess buffer space...")

  // 清理器Cleaner
  val cleaner = new Cleaner(id = threadId,
                            offsetMap = new SkimpyOffsetMap(memory = math.min(config.dedupeBufferSize / config.numThreads, Int.MaxValue).toInt, hashAlgorithm = config.hashAlgorithm),
                            ioBufferSize = config.ioBufferSize / config.numThreads / 2,
                            maxIoBufferSize = config.maxMessageSize,
                            dupBufferLoadFactor = config.dedupeBufferLoadFactor,
                            throttler = throttler,
                            time = time,
                            checkDone = checkDone)
  // 清理线程执行的任务
  override def doWork(): Unit = {
    val cleaned = tryCleanFilthiestLog() // 尝试执行日志压缩
    if (!cleaned)
      pause(config.backOffMs, TimeUnit.MILLISECONDS)
  }
    
  private def tryCleanFilthiestLog(): Boolean = {
    try {
      cleanFilthiestLog()
    } catch {
      // ...
    }
  }

  // 清理/压缩日志
  @throws(classOf[LogCleaningException])
  private def cleanFilthiestLog(): Boolean = {
    val preCleanStats = new PreCleanStats()
    // 选择上面提及的比例最大的日志，进行清理/压缩
    val cleaned = cleanerManager.grabFilthiestCompactedLog(time, preCleanStats) match {
      case None =>
        false
      case Some(cleanable) =>
        this.lastPreCleanStats = preCleanStats
        try {
          cleanLog(cleanable) // 清理/压缩日志
          true
        } catch {
          // ...
        }
    }
    // 删除旧日志分段，分段的偏移量小于日志（不是最新分段）的起始偏移
    val deletable: Iterable[(TopicPartition, Log)] = cleanerManager.deletableLogs()
    try {
      deletable.foreach { case (_, log) =>
        try {
          log.deleteOldSegments() // 删除旧日志分段
        } catch {
          // ...
        }
      }
    } finally  {
      cleanerManager.doneDeleting(deletable.map(_._1)) 
    }
    cleaned
  }
}

private[log] def buildOffsetMap(log: Log,
                                start: Long, // 起始点：日志清理点
                                end: Long, // 结束点：活跃日志分段到基准偏移量
                                map: OffsetMap,
                                stats: CleanerStats): Unit = {
  map.clear()
  val dirty = log.logSegments(start, end).toBuffer
  val nextSegmentStartOffsets = new ListBuffer[Long]
  if (dirty.nonEmpty) {
    for (nextSegment <- dirty.tail) nextSegmentStartOffsets.append(nextSegment.baseOffset)
    nextSegmentStartOffsets.append(end)
  }
  // ...
  // Add all the cleanable dirty segments. We must take at least map.slots * load_factor,
  // but we may be able to fit more (if there is lots of duplication in the dirty section of the log)
  var full = false
  for ( (segment, nextSegmentStartOffset) <- dirty.zip(nextSegmentStartOffsets) if !full) {
    checkDone(log.topicPartition)
    // 扫描所有的分段，构建映射，直到缓存满为止
    full = buildOffsetMapForSegment(log.topicPartition, segment, map, start, nextSegmentStartOffset, log.config.maxMessageSize, transactionMetadata, stats)
    // ...
  }
  info("Offset map for log %s complete.".format(log.name))
}

private def buildOffsetMapForSegment(topicPartition: TopicPartition,
                                     segment: LogSegment,
                                     map: OffsetMap,
                                     startOffset: Long,
                                     nextSegmentStartOffset: Long,
                                     maxLogMessageSize: Int,
                                     transactionMetadata: CleanedTransactionMetadata,
                                     stats: CleanerStats): Boolean = {
  var position = segment.offsetIndex.lookup(startOffset).position
  val maxDesiredMapSize = (map.slots * this.dupBufferLoadFactor).toInt
  while (position < segment.log.sizeInBytes) {
    checkDone(topicPartition)
    readBuffer.clear()
    try {
      // 将分段读入缓存中
      segment.log.readInto(readBuffer, position)
    } catch {
      case e: Exception =>
        throw new KafkaException(s"Failed to read from segment $segment of partition $topicPartition " +
          "while loading offset map", e)
    }
    val records = MemoryRecords.readableRecords(readBuffer)
    throttler.maybeThrottle(records.sizeInBytes)

    val startPosition = position
    // 遍历所有记录
    for (batch <- records.batches.asScala) {
      // ...
      } else {
        val isAborted = transactionMetadata.onBatchRead(batch)
        // ...
        else {
          for (record <- batch.asScala) {
            if (record.hasKey && record.offset >= startOffset) {
              // 构建映射，只需要最大偏移量的记录
              if (map.size < maxDesiredMapSize)
                map.put(record.key, record.offset)
              else
                // 若满了，则直接返回
                return true
            }
            stats.indexMessagesRead(1)
          }
        }
      }
      if (batch.lastOffset >= startOffset)
         map.updateLatestOffset(batch.lastOffset)
    }
    val bytesRead = records.validBytes
    position += bytesRead
    stats.indexBytesRead(bytesRead)

    // if we didn't read even one complete message, our read buffer may be too small
    if(position == startPosition)
      growBuffersOrFail(segment.log, position, maxLogMessageSize, records)
  }
  // In the case of offsets gap, fast forward to latest expected offset in this segment.
  map.updateLatestOffset(nextSegmentStartOffset - 1L)
  restoreBuffers()
  false
}

private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], maxSize: Int, maxIndexSize: Int, firstUncleanableOffset: Long): List[Seq[LogSegment]] = {
  var grouped = List[List[LogSegment]]()
  var segs = segments.toList // 所有参与压缩的日志分段
  while(segs.nonEmpty) {
    var group = List(segs.head) // 每个分组，以第一个分段打底
    var logSize = segs.head.size.toLong
    var indexSize = segs.head.offsetIndex.sizeInBytes.toLong
    var timeIndexSize = segs.head.timeIndex.sizeInBytes.toLong
    segs = segs.tail // 剩余分段
    // 添加分段到分组，直到满了才推出
    while(segs.nonEmpty &&
          logSize + segs.head.size <= maxSize &&
          indexSize + segs.head.offsetIndex.sizeInBytes <= maxIndexSize &&
          timeIndexSize + segs.head.timeIndex.sizeInBytes <= maxIndexSize &&
          lastOffsetForFirstSegment(segs, firstUncleanableOffset) - group.last.baseOffset <= Int.MaxValue) {
      group = segs.head :: group // 追加日志分段到组中
      logSize += segs.head.size
      indexSize += segs.head.offsetIndex.sizeInBytes
      timeIndexSize += segs.head.timeIndex.sizeInBytes
      segs = segs.tail
    }
    grouped ::= group.reverse
  }
  grouped.reverse
}

private[log] def cleanSegments(log: Log,
                               segments: Seq[LogSegment],
                               map: OffsetMap,
                               deleteHorizonMs: Long,
                               stats: CleanerStats,
                               transactionMetadata: CleanedTransactionMetadata): Unit = {
  // 创建一个新分段
  val cleaned = LogCleaner.createNewCleanedSegment(log, segments.head.baseOffset)
  transactionMetadata.cleanedIndex = Some(cleaned.txnIndex)
  try {
    val iter = segments.iterator
    var currentSegmentOpt: Option[LogSegment] = Some(iter.next())
    val lastOffsetOfActiveProducers = log.lastRecordsOfActiveProducers
    // 遍历分组中的每个日志分段，将其清理到新的日志分段中
    // 由于有映射，所以只需要追加即可
    while (currentSegmentOpt.isDefined) {
      val currentSegment = currentSegmentOpt.get
      val nextSegmentOpt = if (iter.hasNext) Some(iter.next()) else None
      val startOffset = currentSegment.baseOffset
      val upperBoundOffset = nextSegmentOpt.map(_.baseOffset).getOrElse(map.latestOffset + 1)
      val abortedTransactions = log.collectAbortedTransactions(startOffset, upperBoundOffset)
      transactionMetadata.addAbortedTransactions(abortedTransactions)
      val retainDeletesAndTxnMarkers = currentSegment.lastModified > deleteHorizonMs
      try {
        // 这里执行清理（压缩）
        cleanInto(log.topicPartition, currentSegment.log, cleaned, map, retainDeletesAndTxnMarkers, log.config.maxMessageSize, transactionMetadata, lastOffsetOfActiveProducers, stats)
      } catch {
        // ...
      }
      currentSegmentOpt = nextSegmentOpt
    }
    // 触发回调，这里是非活跃的（本来就是非活跃的）
    cleaned.onBecomeInactiveSegment()
    // 刷盘
    cleaned.flush()
    // 更新最近更新时间为组中最后一个分段的更新时间
    val modified = segments.last.lastModified
    cleaned.lastModified = modified
    // 用新的压缩过的分段替换旧日志分段
    log.replaceSegments(List(cleaned), segments)
  } catch {
    // ...
  }
}

private[log] def cleanInto(topicPartition: TopicPartition,
                           sourceRecords: FileRecords,
                           dest: LogSegment,
                           map: OffsetMap,
                           retainDeletesAndTxnMarkers: Boolean,
                           maxLogMessageSize: Int,
                           transactionMetadata: CleanedTransactionMetadata,
                           lastRecordsOfActiveProducers: Map[Long, LastRecord],
                           stats: CleanerStats): Unit = {
  val logCleanerFilter: RecordFilter = new RecordFilter {
    var discardBatchRecords: Boolean = _
    // ...
    override def shouldRetainRecord(batch: RecordBatch, record: Record): Boolean = {
      if (discardBatchRecords)
        false
      else
        // 消息过滤的条件
        Cleaner.this.shouldRetainRecord(map, retainDeletesAndTxnMarkers, batch, record, stats)
    }
  }
  var position = 0
  // 扫描分段中的记录
  while (position < sourceRecords.sizeInBytes) {
    checkDone(topicPartition)
    readBuffer.clear()
    writeBuffer.clear()
	  // 先将消息加载到读缓存中
    sourceRecords.readInto(readBuffer, position)
    val records = MemoryRecords.readableRecords(readBuffer)
    throttler.maybeThrottle(records.sizeInBytes)
    // 根据规则过滤记录，得到压缩后的记录
    val result = records.filterTo(topicPartition, logCleanerFilter, writeBuffer, maxLogMessageSize, decompressionBufferSupplier)
    stats.readMessages(result.messagesRead, result.bytesRead)
    stats.recopyMessages(result.messagesRetained, result.bytesRetained)

    position += result.bytesRead
    val outputBuffer = result.outputBuffer
    // 将压缩后的记录写到写缓存中，等待刷盘
    if (outputBuffer.position() > 0) {
      outputBuffer.flip()
      val retained = MemoryRecords.readableRecords(outputBuffer)
      dest.append(largestOffset = result.maxOffset,
        largestTimestamp = result.maxTimestamp,
        shallowOffsetOfMaxTimestamp = result.shallowOffsetOfMaxTimestamp,
        records = retained)
      throttler.maybeThrottle(outputBuffer.limit())
    }
    if (readBuffer.limit() > 0 && result.bytesRead == 0)
      growBuffersOrFail(sourceRecords, position, maxLogMessageSize, records)
  }
  restoreBuffers()
}

// 过滤消息的条件
private def shouldRetainRecord(map: kafka.log.OffsetMap,
                                 retainDeletes: Boolean,
                                 batch: RecordBatch,
                                 record: Record,
                                 stats: CleanerStats): Boolean = {
  // 1. 若消息偏移大于映射中的最大偏移，则保留消息
  val pastLatestOffset = record.offset > map.latestOffset
  if (pastLatestOffset)
    return true

  if (record.hasKey) {
    val key = record.key
    val foundOffset = map.get(key) // 获取映射中的最大偏移
    // 要满足：
    // 1. 消息必须有键
    // 2. 消息的偏移不小于映射中的偏移
    // 3. 消息不能是“删除点”（墓碑标记），且已经过期
    val latestOffsetForKey = record.offset() >= foundOffset
    val isRetainedValue = record.hasValue || retainDeletes // hasValue为true表示不是墓碑标记
    latestOffsetForKey && isRetainedValue
  } else {
    stats.invalidMessage()
    false
  }
}