Apache Kafka 获取因retention-timer到期被标记删除消息的实现方法

原生Kafka未内置单条消息到期删除的回调通知能力，Kafka的消息过期删除是按日志段批量触发的：仅当非活跃日志段内所有消息均满足retention.ms/retention.bytes阈值时，才会整段删除，不存在单条消息标记后立即删除的逻辑，所以要实现单条消息级别的实时删除通知，需要基于扩展能力定制。

• 方案1：自定义Broker端日志保留策略钩子 #

  Kafka 2.0及以上版本支持自定义LogSegmentRetentionPolicy接口实现，你可以扩展该接口重写保留判定逻辑：

  1. 在判定待删除的日志段时，优先读取该段内的所有消息元数据（时间戳、offset、key、分区信息等）
  2. 将需要的消息信息写入你指定的通知专用Kafka Topic，或者直接推送到你的接收组件
  3. 执行原有默认保留策略的删除逻辑即可
     注意：该方案需要修改Broker的配置加载自定义插件，若要实现单条消息粒度的通知，需要额外解析待删除日志段的内容，会增加Broker的IO开销，适合对性能影响容忍度较高的场景

• 方案2：客户端侧旁路消费实现到期监听 #

  无需修改Broker侧配置，完全在客户端实现，测试成本最低，适合好奇验证场景：

  1. 启动一个独立的专属消费者组，消费所有需要监听的Topic消息
  2. 消费时仅记录每条消息的必要信息（key、时间戳、offset、分区）和该Topic配置的retention.ms阈值，存入本地存储或者时序数据库
  3. 启动定时任务扫描到期的消息记录，触发回调通知即可
     注意：该方案需要额外占用消费带宽和元数据存储资源，若Topic保留时间很长、消息量级极高，存储成本会显著上升

• 方案3：替换原生保留策略，业务侧控制过期逻辑 #

  若可调整业务实现，可关闭Topic的原生retention配置，改用业务逻辑控制消息过期：

  1. 所有消息写入时自带业务侧定义的过期时间戳
  2. 用KStream等流处理组件做窗口计算，匹配到期的消息后直接生成通知
  3. 业务侧自行删除已过期的消息，或者统一设置极长的原生保留时间兜底

内容的提问来源于stack exchange，提问作者Sheena