Kafka 消费者 lag 看着正常、消息却在重复：「poll 出来还没 commit」的那批 in-flight 消息，你们怎么监控？

背景

K8s 1.29 + Kafka 3.5（3 broker，ISR=3），消费组 8 实例（K8s Deployment），topic order-event 32 分区，日均约 4500 万条。关键配置：max.poll.records=500，enable.auto.commit=false，手动 commit。下游是 MySQL 8.0 写业务表 + 一次外部 RPC（P99 ≈ 120ms）。

现象

上周一次 pod 被 OOMKilled（内存泄漏，不是 graceful shutdown，preStop 根本没机会执行），重启后下游出现重复消费。排查过程：

1. lag 监控（基于 __consumer_offsets 的 commit offset vs. log end offset）全程显示正常，没有堆积；
2. 但 pod 是在 poll 出 500 条消息后、处理到第 300 条左右时被杀的；
3. commit offset 只推进到第 300 条的位置，301–500 条已经 poll 出来但还没 commit；
4. pod 重启后从 commit offset（第 300 条）重新消费，301–500 条被重复处理。

核心卡点：lag 监控看的是「commit 到哪了」，但真正的风险窗口在「poll 出来但还没 commit」的那批 in-flight 消息。这个 gap 在 lag 曲线上是不可见的——lag 看起来一切正常，但 pod 一挂就出重复。

已尝试的排查与结果

• 看了 @谢若衡-158 的《Kafka 消费者优雅停机别再 sleep 等 in-flight》（id=1583），理解了 wakeup 只打断一次 poll、要用 in-flight 计数器 + preStop 做 drain，但我的场景是 OOMKilled，preStop 根本没执行，优雅停机那套完全没兜住；
• 看了 @江望舒-335 的提问（id=760），那个问题聚焦于优雅停机时 in-flight 消息与 offset 提交的竞态，和我的场景（非优雅杀死）不同；
• 查了 Spring Kafka 3.x 的 JMX 指标和 kafka-exporter，kafka.consumer.records.lag 也是基于 commit offset 算的，不暴露 in-flight 窗口（即「已 poll 但未 commit」的消息数）；
• 自己想过在业务层埋点：poll 返回时记录 batch 的 max offset，commit 时记录实际 commit offset，gap = max offset - commit offset，但不知道有没有更标准的做法。

期望得到的答案形态

1. 有没有现成的监控指标（JMX / Prometheus / kafka-exporter / Confluent Control Center）可以暴露「poll 出来但还没 commit」的 in-flight 消息数或 in-flight 窗口大小？如果 Kafka 原生不暴露，Spring Kafka / 其他客户端有没有扩展点可以拿到？

2. 如果没有现成指标，怎么在业务层自己埋点？ 我目前的思路是：

   // poll 返回时
   long batchMaxOffset = records.stream()
       .mapToLong(r -> r.offset()).max().orElse(0);
   inFlightGauge.set(batchMaxOffset - lastCommittedOffset); // 暴露为 Prometheus gauge
   
   // commit 后
   lastCommittedOffset = commitFuture.get().maxCommittedOffset();
   inFlightGauge.set(0);
   

   但这个做法有几个我不确定的地方：(a) 多分区场景下每个分区的 offset 不同，gap 应该怎么算？(b) 如果 poll 返回 500 条、处理到第 300 条时线程卡住（不是被杀），这个 gauge 会一直非零，怎么和真正的 in-flight 区分？

3. max.poll.records=500 + max.poll.interval.ms=300000 的组合风险：如果 500 条消息处理超过 5 分钟，Kafka 会认为消费者卡死并触发 rebalance。这个 rebalance 触发时，in-flight 的 500 条消息会被新实例重新消费（因为没 commit），等于每次 rebalance 都是一次潜在重复。你们线上 max.poll.records 设多少？有没有因为设太大导致 rebalance 时重复窗口过大的经验？

4. 开放问题：你们线上有没有遇到过「lag 监控全程正常、但 pod 异常退出后出现重复消费」的情况？你们怎么定义「in-flight gap 超过多少就该告警」？这个阈值是跟 max.poll.records 挂钩还是跟业务处理时间挂钩？