Kafka delete 日志清理策略（日志删除）

• 原理：按照一定保留策略，直接删除不符合条件的日志分段。Kafka 把 topic 的一个 partition 大文件分成多个小文件段，通过这种方式，能方便地定期清除或删除已消费完的文件，以减少磁盘占用 。
• 保留策略
  • 按时间删除：设定一个时间阈值，删除修改时间在该时间之前的日志。比如设置log.retention.hours = 1 ，就表示只保存 1 小时内的日志，超出 1 小时的日志分段会被删除。
  • 按大小删除：指定一个数据大小阈值，当日志数据超过这个大小时，执行删除操作，保留最后的指定大小数据。例如log.retention.bytes = 1073741824 （即 1GB ），表示日志数据超过 1GB 时，会删除旧的消息。
• 相关参数
  • log.cleanup.policy = delete ：启用删除策略。
  • log.retention.check.interval.ms ：专门的日志删除任务周期性检测的时间间隔，默认 300000ms（5 分钟 ），即每隔 5 分钟检查一次是否有符合删除条件的日志分段 。
  • log.retention.hours ：按时间清理时，指定日志保留的小时数 。
  • log.retention.bytes ：按大小清理时，指定日志保留的字节数 。需注意log.retention.bytes和log.retention.hours任意一个达到要求，都会执行删除操作，且会被 topic 创建时的指定参数覆盖 。

Kafka compact 日志清理策略（日志压缩）

• 原理：针对每个消息的 key 进行整合，对于有相同 key 的不同 value 值，只保留最后一个版本 。就像在一个记录集合里，相同标识（key ）的记录，只留下最新的那条 。比如一个 key 对应的值先后为 “值 1”“值 2”“值 3” ，经过日志压缩后，只会保留 “值 3” 。清理重复 key 后，一些 segment 文件大小会变小，Kafka 会将小文件再合并成大的 segment 文件 。
• 特殊处理：当某个 key 的最新版本消息没有内容（value 为 null ）时，这个 key 将被删除，这类消息被称为 “墓碑消息（tombstone ）” 。墓碑消息的存放时间和 broker 的配置log.cleaner.delete.retention.ms有关，默认值是 24 小时 。在执行日志清理时，会删除到期的墓碑消息 。
• 清理流程
  • 对于每个 Kafka partition 的日志，以 segment 为单位，分为已清理和未清理部分，未清理部分又细分为可清理和不可清理的。
  • 每个日志目录下有cleaner - offset - checkpoint文件记录清理进度。
  • 找出可清理的 segment（active segment 不能清理，根据min.compaction.lag.ms配置判断其他 segment 是否能清理 ，即判断 segment 最后一条记录的插入时间是否超过最小保留时间 ）。
  • 构建 SkimpyOffsetMap 对象（key 与 offset 的映射哈希表 ），遍历可清理 segment 的每条日志，将 key 和 offset 存入其中。
  • 再遍历已清理和可清理部分的 segment 日志，根据 SkimpyOffsetMap 判断是否保留日志 。
  • 执行清理操作，将可清理部分的 segment 变为已清理的，同时更新 cleaner checkpoint 记录的 offset 。
• 相关参数
  • log.cleaner.enable = true ：开启日志压缩功能。
  • log.cleanup.policy = compact ：启用日志压缩策略 。
  • min.compaction.lag.ms ：设置日志段中消息可被压缩的最小时间间隔 ，确保消息存留一定时间后才可能被清理 。
  • log.cleaner.delete.retention.ms ：墓碑消息的保留时长 。