• 這里說的日志,是指Kafka保存寫入消息的文件;
• Kafka日志清除策略包括中間:
  1. 基于時間和大小的刪除策略;
  2. Compact清理策略;
• 我們這里主要介紹基于Compact策略的Log Clean;

Compact策略說明

• Kafka官網(wǎng)介紹: Log compaction;
• Compact就是壓縮, 只能針對特定的topic應(yīng)用此策略,即寫入的message都帶有Key, 合并相同Key的message, 只留下最新的message;
• 在壓縮過程中, 針對message的payload為null的也將會去除掉;
• 官網(wǎng)上扒了一張圖, 大家先感受下:

110.png

日志清理過程中的狀態(tài)

• 主要涉及三種狀態(tài): LogCleaningInProgress, LogCleaningAborted,和LogCleaningPaused, 從字面上就很容易理解是什么意思,下面是源碼中的注釋:

• If a partition is to be cleaned, it enters the LogCleaningInProgress state.
• While a partition is being cleaned, it can be requested to be aborted and paused. Then the partition first enters
• the LogCleaningAborted state. Once the cleaning task is aborted, the partition enters the LogCleaningPaused state.
• While a partition is in the LogCleaningPaused state, it won't be scheduled for cleaning again, until cleaning is requested to be resumed.

• LogCleanerManager類 管理所有清理的log的狀態(tài)及轉(zhuǎn)換:

def abortCleaning(topicAndPartition: TopicAndPartition)
def abortAndPauseCleaning(topicAndPartition: TopicAndPartition)
def resumeCleaning(topicAndPartition: TopicAndPartition)
def checkCleaningAborted(topicAndPartition: TopicAndPartition) 

要清理的日志的選取

• 因?yàn)檫@個compact清理過程涉及到log和index等文件的重寫,比較耗IO, 因此kafka會作流控, 每次compact時都會先按規(guī)則確定要清理哪些TopicAndPartiton的log;
• 使用LogToClean類來表示要被清理的Log:

private case class LogToClean(topicPartition: TopicAndPartition, log: Log, firstDirtyOffset: Long) extends Ordered[LogToClean] {
  val cleanBytes = log.logSegments(-1, firstDirtyOffset).map(_.size).sum
  val dirtyBytes = log.logSegments(firstDirtyOffset, math.max(firstDirtyOffset, log.activeSegment.baseOffset)).map(_.size).sum
  val cleanableRatio = dirtyBytes / totalBytes.toDouble
  def totalBytes = cleanBytes + dirtyBytes
  override def compare(that: LogToClean): Int = math.signum(this.cleanableRatio - that.cleanableRatio).toInt
}

1. firstDirtyOffset:表示本次清理的起始點(diǎn), 其前邊的offset將被作清理,與在其后的message作key的合并;
2. val cleanableRatio = dirtyBytes / totalBytes.toDouble, 需要清理的log的比例,這個值越大,越可能被最后選中作清理;
3. 每次清理完,要更新當(dāng)前已經(jīng)清理到的位置, 記錄在cleaner-offset-checkpoint文件中,作為下一次清理時生成firstDirtyOffset的參考;

def updateCheckpoints(dataDir: File, update: Option[(TopicAndPartition,Long)]) {
    inLock(lock) {
      val checkpoint = checkpoints(dataDir)
      val existing = checkpoint.read().filterKeys(logs.keys) ++ update
      checkpoint.write(existing)
    }
  }

• 選出最需要清理的日志:

def grabFilthiestLog(): Option[LogToClean] = {
    inLock(lock) {
      val lastClean = allCleanerCheckpoints()
      val dirtyLogs = logs.filter {
        case (topicAndPartition, log) => log.config.compact  // skip any logs marked for delete rather than dedupe
      }.filterNot {
        case (topicAndPartition, log) => inProgress.contains(topicAndPartition) // skip any logs already in-progress
      }.map {
        case (topicAndPartition, log) => // create a LogToClean instance for each
          // if the log segments are abnormally truncated and hence the checkpointed offset
          // is no longer valid, reset to the log starting offset and log the error event
          val logStartOffset = log.logSegments.head.baseOffset
          val firstDirtyOffset = {
            val offset = lastClean.getOrElse(topicAndPartition, logStartOffset)
            if (offset < logStartOffset) {
              error("Resetting first dirty offset to log start offset %d since the checkpointed offset %d is invalid."
                    .format(logStartOffset, offset))
              logStartOffset
            } else {
              offset
            }
          }
          LogToClean(topicAndPartition, log, firstDirtyOffset)
      }.filter(ltc => ltc.totalBytes > 0) // skip any empty logs

      this.dirtiestLogCleanableRatio = if (!dirtyLogs.isEmpty) dirtyLogs.max.cleanableRatio else 0
      // and must meet the minimum threshold for dirty byte ratio
      val cleanableLogs = dirtyLogs.filter(ltc => ltc.cleanableRatio > ltc.log.config.minCleanableRatio)
      if(cleanableLogs.isEmpty) {
        None
      } else {
        val filthiest = cleanableLogs.max
        inProgress.put(filthiest.topicPartition, LogCleaningInProgress)
        Some(filthiest)
      }
    }
  }

代碼看著多,實(shí)在比較簡單:

1. 從所有的Log中產(chǎn)生出 LogToClean對象列表;
2. 從1中獲得的LogToClean列表中過濾過cleanableRatio大于config中配置的清理比率的LogToClean;
3. 從2中獲取的LogToClean列表中取cleanableRatio最大的,即為當(dāng)前最需要被清理的.

先放兩張網(wǎng)上扒來的圖:

111.png

1. 這里的CleanerPoint就是我們上面說的firstDirtyOffset;
2. Log Tail中的key將被合并到 LogHead中,實(shí)際上因?yàn)闃?gòu)建OffsetMap是在Log Head部分,因此合并Key的部分還包括構(gòu)建OffsetMap最后到達(dá)的Offset位置;

下面這個是整個壓縮合并的過程, Kafka的代碼就是把這個過程翻譯成Code

112.png

構(gòu)建OffsetMap

• 構(gòu)建上面圖111.png中LogHead部分的所有日志的OffsetMap, 此Map中的key即為message.key的hash值, value即為當(dāng)前message的offset
• 實(shí)現(xiàn):

private[log] def buildOffsetMap(log: Log, start: Long, end: Long, map: OffsetMap): Long = {
    map.clear()
    val dirty = log.logSegments(start, end).toSeq
    info("Building offset map for log %s for %d segments in offset range [%d, %d).".format(log.name, dirty.size, start, end))
    
    // Add all the 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 offset = dirty.head.baseOffset
    require(offset == start, "Last clean offset is %d but segment base offset is %d for log %s.".format(start, offset, log.name))
    val maxDesiredMapSize = (map.slots * this.dupBufferLoadFactor).toInt
    var full = false
    for (segment <- dirty if !full) {
      checkDone(log.topicAndPartition)
      val segmentSize = segment.nextOffset() - segment.baseOffset

      require(segmentSize <= maxDesiredMapSize, "%d messages in segment %s/%s but offset map can fit only %d. You can increase log.cleaner.dedupe.buffer.size or decrease log.cleaner.threads".format(segmentSize,  log.name, segment.log.file.getName, maxDesiredMapSize))
      if (map.size + segmentSize <= maxDesiredMapSize)
        offset = buildOffsetMapForSegment(log.topicAndPartition, segment, map)
      else
        full = true
    }
    info("Offset map for log %s complete.".format(log.name))
    offset
  }

1. 順序讀取每個LogSegment, 將相關(guān)信息put到OffsetMap, 其中的key是message.key的hash值, 這個地方有個坑,如果出現(xiàn)了hash碰撞怎么?
2. build的OffsetMap有大小限制, 不能超過val maxDesiredMapSize = (map.slots * this.dupBufferLoadFactor).toInt.

重新分組需要清理的LogSegments

• 因?yàn)閴嚎s清理后,原來的單個LogSegment勢必大小要減少,因此需要重新分組來為重寫Log和Index文件作準(zhǔn)備;
• 分組的規(guī)則也很簡單: 根據(jù)segmentsize和indexsize進(jìn)行分組,這個分組是每一組的segmentsize不能超過segmentSize的配置大小,indexfile不能超過配置的最大indexsize的大小,同時條數(shù)不能超過int.maxvalue.

private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], maxSize: Int, maxIndexSize: Int): List[Seq[LogSegment]] = {
    var grouped = List[List[LogSegment]]()
    var segs = segments.toList
    while(!segs.isEmpty) {
      var group = List(segs.head)
      var logSize = segs.head.size
      var indexSize = segs.head.index.sizeInBytes
      segs = segs.tail
      while(!segs.isEmpty &&
            logSize + segs.head.size <= maxSize &&
            indexSize + segs.head.index.sizeInBytes <= maxIndexSize &&
            segs.head.index.lastOffset - group.last.index.baseOffset <= Int.MaxValue) {
        group = segs.head :: group
        logSize += segs.head.size
        indexSize += segs.head.index.sizeInBytes
        segs = segs.tail
      }
      grouped ::= group.reverse
    }
    grouped.reverse
  }

按上面重新分成的組作真正的清理工作

• 清理的過程,遍歷所有需要清理的LogSegment, 按一定的規(guī)則過濾出需要保留的msg重定入新的Log文件中;
• 符合下列規(guī)則的message將被保留
  1. message的key在OffsetMap中能找到,同時當(dāng)前的message的offset不小于offsetMap中存儲的offset;
  2. 這個segment的最后修改時間大于最大的保留時間,同時這個消息的value是有效的value,即不為null;

private def shouldRetainMessage(source: kafka.log.LogSegment,
                                  map: kafka.log.OffsetMap,
                                  retainDeletes: Boolean,
                                  entry: kafka.message.MessageAndOffset): Boolean = {
    val key = entry.message.key
    if (key != null) {
      val foundOffset = map.get(key)
      /* two cases in which we can get rid of a message:
       *   1) if there exists a message with the same key but higher offset
       *   2) if the message is a delete "tombstone" marker and enough time has passed
       */
      val redundant = foundOffset >= 0 && entry.offset < foundOffset
      val obsoleteDelete = !retainDeletes && entry.message.isNull
      !redundant && !obsoleteDelete
    } else {
      stats.invalidMessage()
      false
    }
  }

• 清理:
  1. 實(shí)現(xiàn)下就是按上面的規(guī)則過濾過需要保留的message后重寫Log和Index文件過程;
  2. 具體寫文件的流程可參考 Kafka中Message存儲相關(guān)類大揭密

private[log] def cleanSegments(log: Log,
                                 segments: Seq[LogSegment], 
                                 map: OffsetMap, 
                                 deleteHorizonMs: Long) {
    // create a new segment with the suffix .cleaned appended to both the log and index name
    val logFile = new File(segments.head.log.file.getPath + Log.CleanedFileSuffix)
    logFile.delete()
    val indexFile = new File(segments.head.index.file.getPath + Log.CleanedFileSuffix)
    indexFile.delete()
    val messages = new FileMessageSet(logFile, fileAlreadyExists = false, initFileSize = log.initFileSize(), preallocate = log.config.preallocate)
    val index = new OffsetIndex(indexFile, segments.head.baseOffset, segments.head.index.maxIndexSize)
    val cleaned = new LogSegment(messages, index, segments.head.baseOffset, segments.head.indexIntervalBytes, log.config.randomSegmentJitter, time)

    try {
      // clean segments into the new destination segment
      for (old <- segments) {
        val retainDeletes = old.lastModified > deleteHorizonMs
        info("Cleaning segment %s in log %s (last modified %s) into %s, %s deletes."
            .format(old.baseOffset, log.name, new Date(old.lastModified), cleaned.baseOffset, if(retainDeletes) "retaining" else "discarding"))
        cleanInto(log.topicAndPartition, old, cleaned, map, retainDeletes)
      }

      // trim excess index
      index.trimToValidSize()

      // flush new segment to disk before swap
      cleaned.flush()

      // update the modification date to retain the last modified date of the original files
      val modified = segments.last.lastModified
      cleaned.lastModified = modified

      // swap in new segment
      info("Swapping in cleaned segment %d for segment(s) %s in log %s.".format(cleaned.baseOffset, segments.map(_.baseOffset).mkString(","), log.name))
      log.replaceSegments(cleaned, segments)
    } catch {
      case e: LogCleaningAbortedException =>
        cleaned.delete()
        throw e
    }
  }

Kafka源碼分析-匯總