重磅好文！现场故障回顾：最新版Kafka数据过期未删除问题的源码深入剖析与终极解决方案总结

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"

try server.startup()

 /* start scheduler */ kafkaScheduler = new KafkaScheduler(config.backgroundThreads) kafkaScheduler.startup()

        /* start log manager */
        _logManager = LogManager(
          config,
          metaPropsEnsemble.errorLogDirs().asScala.toSeq,
          configRepository,
          kafkaScheduler,
          time,
          brokerTopicStats,
          logDirFailureChannel,
          config.usesTopicId)
        _brokerState = BrokerState.RECOVERY
        logManager.startup(zkClient.getAllTopicsInCluster())

 /* 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(),
                         InitialTaskDelayMs,
                         retentionCheckMs)
      info("Starting log flusher with a default period of %d ms.".format(flushCheckMs))
      scheduler.schedule("kafka-log-flusher",
                         () => flushDirtyLogs(),
                         InitialTaskDelayMs,
                         flushCheckMs)
      scheduler.schedule("kafka-recovery-point-checkpoint",
                         () => checkpointLogRecoveryOffsets(),
                         InitialTaskDelayMs,
                         flushRecoveryOffsetCheckpointMs)
      scheduler.schedule("kafka-log-start-offset-checkpoint",
                         () => checkpointLogStartOffsets(),
                         InitialTaskDelayMs,
                         flushStartOffsetCheckpointMs)
      scheduler.scheduleOnce("kafka-delete-logs", // will be rescheduled after each delete logs with a dynamic period
                         () => deleteLogs(),
                         InitialTaskDelayMs)
}

 deletableLogs.foreach {
        case (topicPartition, log) =>
          debug(s"Garbage collecting '${log.name}'")
          total += log.deleteOldSegments()

          val futureLog = futureLogs.get(topicPartition)
          if (futureLog != null) {
            // clean future logs
            debug(s"Garbage collecting future log '${futureLog.name}'")
            total += futureLog.deleteOldSegments()
          }
      }

  /**
   * If topic deletion is enabled, delete any local log segments that have either expired due to time based retention
   * or because the log size is > retentionSize.
   *
   * Whether or not deletion is enabled, delete any local log segments that are before the log start offset
   */
  def deleteOldSegments(): Int = {
    if (config.delete) {
      deleteLogStartOffsetBreachedSegments() +
        deleteRetentionSizeBreachedSegments() +
        deleteRetentionMsBreachedSegments()
    } else {
      deleteLogStartOffsetBreachedSegments()
    }
  }

    private def deleteRetentionMsBreachedSegments(): Int = {
    val retentionMs = localRetentionMs(config, remoteLogEnabled())
    if (retentionMs < 0) return 0
    val startMs = time.milliseconds

    def shouldDelete(segment: LogSegment, nextSegmentOpt: Option[LogSegment]): Boolean = {
      startMs - segment.largestTimestamp > retentionMs
    }

    deleteOldSegments(shouldDelete, RetentionMsBreach(this, remoteLogEnabled()))
  }

    /**
     * The largest timestamp this segment contains.
     */
    public long largestTimestamp() throws IOException {
        long maxTimestampSoFar = maxTimestampSoFar();
        if (maxTimestampSoFar >= 0)
            return maxTimestampSoFar;
        return lastModified();
    }

log.retention.minutes=1
log.retention.check.interval.ms=1000

     /**
     * read byte array from input stream and then generate an iterator of producer record
     * @param {@link InputStream} of messages. the implementation does not need to close the input stream.
     * @return an iterator of producer record. It should implement following rules. 1) the hasNext() method must be idempotent.
     *         2) the convert error should be thrown by next() method.
     */
Iterator<ProducerRecord<byte[], byte[]>> readRecords(InputStream inputStream);

https://github.com/felixzh2020/felixzh-learning-kafka/tree/master/TimeStampCase/src/main/java/com/felixzh/learning

ProducerRecordproducerRecordWithOutTimeStamp = new ProducerRecord<>("test", "hello felixzh");

ProducerRecordproducerRecordWithTimeStamp = new ProducerRecord<>("test", null, 1000L, null, "hello felixzh");

ProducerRecordproducerRecordWithTimeStamp = new ProducerRecord<>("test", null, 1742299840000L, null, "hello felixzh");

CreateTime表示使用prodcuer写数据时携带的timestamp；
LogAppendTime表示使用Kafka写文件时候的系统timestamp；

long timestamp = record.timestamp() == null ? nowMs : record.timestamp();      

kafka\core\src\main\scala\kafka\log\UnifiedLog.scala

 /**
   * Append this message set to the active segment of the local log, rolling over to a fresh segment if necessary.
   */
  private def append(records: MemoryRecords,
                     origin: AppendOrigin,
                     interBrokerProtocolVersion: MetadataVersion,
                     validateAndAssignOffsets: Boolean,
                     leaderEpoch: Int,
                     requestLocal: Option[RequestLocal],
                     verificationGuard: VerificationGuard,
                     ignoreRecordSize: Boolean): LogAppendInfo

public ValidationResult validateMessagesAndAssignOffsets(PrimitiveRef.LongRef offsetCounter,
                                                             MetricsRecorder metricsRecorder,
                                                             BufferSupplier bufferSupplier) {
        if (sourceCompression == CompressionType.NONE && targetCompression == CompressionType.NONE) {
            // check the magic value
            if (!records.hasMatchingMagic(toMagic))
                return convertAndAssignOffsetsNonCompressed(offsetCounter, metricsRecorder);
            else
                // Do in-place validation, offset assignment and maybe set timestamp
                return assignOffsetsNonCompressed(offsetCounter, metricsRecorder);
        } else
            return validateMessagesAndAssignOffsetsCompressed(offsetCounter, metricsRecorder, bufferSupplier);    
    }

long maxTimestamp = RecordBatch.NO_TIMESTAMP;
if (record.timestamp() > maxTimestamp)
       maxTimestamp = record.timestamp();
if (timestampType == TimestampType.LOG_APPEND_TIME)
                maxTimestamp = now;

/home/myHadoopCluster/kafka_2.12-3.7.0/bin/kafka-dump-log.sh --files 00000000000000000005.log

./kafka-console-consumer.sh --bootstrap-server felixzh:9092 --topic test --from-beginning --property print.timestamp=true