深入理解Kafka服务端之日志对象对日志段及关键偏移量的管理

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

def addSegment(segment: LogSegment): LogSegment = this.segments.put(segment.baseOffset, segment)

private def deleteRetentionMsBreachedSegments(): Int = {
  if (config.retentionMs < 0) return 0
  //获取当前时间
  val startMs = time.milliseconds
  //删除不满足保留策略的日志段，返回删除的日志段数量
  //这里的不满足的日志段指：当前时间 - 日志段的最大时间戳 = 日志段未更改的持续时间（保留时间）
  //持续时间 > config.retentionMs（默认7天） ，则删除
  deleteOldSegments((segment, _) => startMs - segment.largestTimestamp > config.retentionMs,
    reason = s"retention time ${config.retentionMs}ms breach")
}

private def deleteRetentionSizeBreachedSegments(): Int = {
  if (config.retentionSize < 0 || size < config.retentionSize) return 0
  //计算要删除的日志的字节数：diff = size（日志的总字节数）- config.retentionSize（日志大小阈值，默认为-1，即无穷大）
  var diff = size - config.retentionSize
  //判断是否需要删除的函数
  def shouldDelete(segment: LogSegment, nextSegmentOpt: Option[LogSegment]) = {
    //如果需要删除的字节 - 日志段的字节 > 0 说明需要被删除
    if (diff - segment.size >= 0) {
      diff -= segment.size
      true
    } else {
      false
    }
  }
  //执行删除
  deleteOldSegments(shouldDelete, reason = s"retention size in bytes ${config.retentionSize} breach")
}

private def deleteLogStartOffsetBreachedSegments(): Int = {
  //定义是否需要删除给定的日志段
  def shouldDelete(segment: LogSegment, nextSegmentOpt: Option[LogSegment]) =
    //如果给定日志段的下一个日志段的起始偏移量 <= LogStartOffset ，则删除
    nextSegmentOpt.exists(_.baseOffset <= logStartOffset)


  deleteOldSegments(shouldDelete, reason = s"log start offset $logStartOffset breach")
}

private def deleteOldSegments(predicate: (LogSegment, Option[LogSegment]) => Boolean, reason: String): Int = {
  lock synchronized {
    //获取可删除的日志段集合
    val deletable = deletableSegments(predicate)
    if (deletable.nonEmpty)
      info(s"Found deletable segments with base offsets [${deletable.map(_.baseOffset).mkString(",")}] due to $reason")
    //删除日志段文件
    deleteSegments(deletable)
  }
}

private def deletableSegments(predicate: (LogSegment, Option[LogSegment]) => Boolean): Iterable[LogSegment] = {
  //如果segments集合为空 或者 高水位为空 则返回空序列
  if (segments.isEmpty || replicaHighWatermark.isEmpty) {
    Seq.empty
  } else {
    //获取高水位值
    val highWatermark = replicaHighWatermark.get
    //定义保存可删除日志段的数组
    val deletable = ArrayBuffer.empty[LogSegment]
    //获取segments集合中的第一个key-value
    var segmentEntry = segments.firstEntry
    // 从具有最小起始位移值的日志段对象开始遍历，直到满足以下条件之一便停止遍历：
    // 1. 测定条件函数predicate = false
    // 2. 扫描到包含Log对象高水位值所在的日志段对象
    // 3. 最新的日志段对象不包含任何消息
    // 最新日志段对象是segments中Key值最大对应的那个日志段，也就是我们常说的Active Segment。完全为空的Active Segment如果被允许删除，后面还要重建它
    // 故代码这里不允许删除大小为空的Active Segment。
    // 在遍历过程中，同时不满足以上3个条件的所有日志段都是可以被删除的！
    while (segmentEntry != null) {
      //获取第一个键值对的value值，即LogSegment对象
      val segment = segmentEntry.getValue
      //如果存在key比segment的key还大的Entry
      val nextSegmentEntry = segments.higherEntry(segmentEntry.getKey)
      //根据是否存在key比segment的key还大的Entry，定义三元组
      val (nextSegment, upperBoundOffset, isLastSegmentAndEmpty) = if (nextSegmentEntry != null)
        //如果存在
        (nextSegmentEntry.getValue, nextSegmentEntry.getValue.baseOffset, false)
      else
        //不存在
        (null, logEndOffset, segment.size == 0)


      if (highWatermark >= upperBoundOffset && predicate(segment, Option(nextSegment)) && !isLastSegmentAndEmpty) {
        //将日志段添加到可删除数组
        deletable += segment
        //更新 segmentEntry 变量
        segmentEntry = nextSegmentEntry
      } else {
        segmentEntry = null
      }
    }
    //返回可删除的日志段数组
    deletable
  }
}

private def deleteSegments(deletable: Iterable[LogSegment]): Int = {
  maybeHandleIOException(s"Error while deleting segments for $topicPartition in dir ${dir.getParent}") {
    //要删除的日志段数量
    val numToDelete = deletable.size
    if (numToDelete > 0) {
      // 如果需要删除所有的日志段文件，则需要先创建一个空的日志段
      if (segments.size == numToDelete)
        //滚动生成新的日志段
        roll()
      lock synchronized {
        checkIfMemoryMappedBufferClosed()
        //便利可删除日志段数组，依次删除日志段
        deletable.foreach(deleteSegment)
        //是否更新Log日志的起始偏移量
        maybeIncrementLogStartOffset(segments.firstEntry.getValue.baseOffset)
      }
    }
    numToDelete
  }

private def deleteSegment(segment: LogSegment) {
  info(s"Scheduling log segment [baseOffset ${segment.baseOffset}, size ${segment.size}] for deletion.")
  lock synchronized {
    //将日志段对象从跳表集合中移除
    segments.remove(segment.baseOffset)
    //异步删除
    asyncDeleteSegment(segment)
  }
}

def logSegments(from: Long, to: Long): Iterable[LogSegment] = {
  lock synchronized {
    val view = Option(segments.floorKey(from)).map { floor =>
      if (to < floor)
        throw new IllegalArgumentException(s"Invalid log segment range: requested segments in $topicPartition " +
          s"from offset $from mapping to segment with base offset $floor, which is greater than limit offset $to")
      segments.subMap(floor, to)
    }.getOrElse(segments.headMap(to))
    view.values.asScala
  }
}

@volatile private var nextOffsetMetadata: LogOffsetMetadata = _

locally {
  ...
  //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)


  ...
}

private def updateLogEndOffset(messageOffset: Long) {
  nextOffsetMetadata = new LogOffsetMetadata(messageOffset, activeSegment.baseOffset, activeSegment.size)
}

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

def maybeIncrementLogStartOffset(newLogStartOffset: Long) {
  // We don't have to write the log start offset to log-start-offset-checkpoint immediately.
  // The deleteRecordsOffset may be lost only if all in-sync replicas of this broker are shutdown
  // in an unclean manner within log.flush.start.offset.checkpoint.interval.ms. The chance of this happening is low.
  maybeHandleIOException(s"Exception while increasing log start offset for $topicPartition to $newLogStartOffset in dir ${dir.getParent}") {
    lock synchronized {
      checkIfMemoryMappedBufferClosed()
      if (newLogStartOffset > logStartOffset) {
        info(s"Incrementing log start offset to $newLogStartOffset")
        logStartOffset = newLogStartOffset
        leaderEpochCache.foreach(_.truncateFromStart(logStartOffset))
        producerStateManager.truncateHead(logStartOffset)
        updateFirstUnstableOffset()
      }
    }
  }
}