问题描述

1.问题定位

最近需要利用flume采集神策的历史数据，数据量比较大，每天大概有2000万条数据，大概要采集一个月的。然后发现数据还没来得及消费就被删除了，

2.问题处理

查看kafka配置后，发现两个关于删除日志策略的参数都设置了。
log.retention.hours和log.retention.bytes
1. 查看kafka0.9.0.0源码，LogManager类

   def cleanupLogs() {
    debug("Beginning log cleanup...")
    var total = 0
    val startMs = time.milliseconds
    for(log <- allLogs; if !log.config.compact) {
      debug("Garbage collecting '" + log.name + "'")
      total += cleanupExpiredSegments(log) + cleanupSegmentsToMaintainSize(log)
    }
    debug("Log cleanup completed. " + total + " files deleted in " +
                  (time.milliseconds - startMs) / 1000 + " seconds")
  }

2. 查看total += cleanupExpiredSegments(log) + cleanupSegmentsToMaintainSize(log)两个方法，删除逻辑是或
首先查看cleanupExpiredSegments(log) 方法，

 private def cleanupExpiredSegments(log: Log): Int = {
    if (log.config.retentionMs < 0)
      return 0
    val startMs = time.milliseconds
    log.deleteOldSegments(startMs - _.lastModified > log.config.retentionMs)
  }```
 startMs 这个参数的源码查看在这里
```scala
private def getLogRetentionTimeMillis: Long = {
    val millisInMinute = 60L * 1000L
    val millisInHour = 60L * millisInMinute

    val millis: java.lang.Long =
      Option(getLong(KafkaConfig.LogRetentionTimeMillisProp)).getOrElse(
        Option(getInt(KafkaConfig.LogRetentionTimeMinutesProp)) match {
          case Some(mins) => millisInMinute * mins
          case None => getInt(KafkaConfig.LogRetentionTimeHoursProp) * millisInHour
        })

    if (millis < 0) return -1
    millis   //总之就是 log.retention.hours log.retention.minutes log.retention.ms配置文件这三个来决定的，把他们转为毫秒
  }

所以cleanupExpiredSegments(log) 方法会根据日志超时时间来删除

查看cleanupSegmentsToMaintainSize(log)方法

private def cleanupSegmentsToMaintainSize(log: Log): Int = {
    if(log.config.retentionSize < 0 || log.size < log.config.retentionSize)
      return 0
    var diff = log.size - log.config.retentionSize
    def shouldDelete(segment: LogSegment) = {
      if(diff - segment.size >= 0) {
        diff -= segment.size
        true
      } else {
        false
      }
    }
    log.deleteOldSegments(shouldDelete)
  }

根据日志大小来决定，这个参数是log.retention.bytes
总结：kafka删除日志的逻辑是通过时间和日志大小来删除的，配置参数log.retention.hours和log.retention.bytes,如果只要按时间删除，则不设置log.retention.bytes，默认是-1，今天的问题就在于两个参数都设置了，所以出现日志过大就被删除了。

3.最后

Kafka日志是多久删除一次的呢？
LogManager类中的startup

def startup() {
    /* Schedule the cleanup task to delete old logs */
    if(scheduler != null) {
      info("Starting log cleanup with a period of %d ms.".format(retentionCheckMs))
      scheduler.schedule("kafka-log-retention", 
                         cleanupLogs, 
                         delay = InitialTaskDelayMs, 
                         period = retentionCheckMs, 
                         TimeUnit.MILLISECONDS)
      info("Starting log flusher with a default period of %d ms.".format(flushCheckMs))
      scheduler.schedule("kafka-log-flusher", 
                         flushDirtyLogs, 
                         delay = InitialTaskDelayMs, 
                         period = flushCheckMs, 
                         TimeUnit.MILLISECONDS)
      scheduler.schedule("kafka-recovery-point-checkpoint",
                         checkpointRecoveryPointOffsets,
                         delay = InitialTaskDelayMs,
                         period = flushCheckpointMs,
                         TimeUnit.MILLISECONDS)
    }
    if(cleanerConfig.enableCleaner)
      cleaner.startup()
  }```
寻找` retentionCheckMs,`是由这个参数决定的，具体过程不详细秒速，`log.retention.check.interval.ms`，默认是300秒，就是5分钟执行清理逻辑一次。