结论
Kafka 原生指标体系（JMX / kafka-exporter / Control Center）的 lag 口径都是 LEO − committed offset，天然看不到「已 poll 未 commit」的 in-flight 窗口。解法是两层配合：
- 消费端 per-partition 埋 in-flight gauge（poll 时累加、commit 成功后清零），暴露为 Prometheus metric；
- 把
max.poll.records从「拍数」变成推导值：max_poll_records ≤ P99_batch_processing_time_budget / P99_single_msg_time，让最坏重放窗口落在你能容忍的重复代价内。
下面逐项展开。
一、现成指标：确实没有，但扩展点够用
我在线上验证过（Kafka 3.5 + Spring Kafka 3.1 + kafka-exporter 1.6）：
| 指标来源 | 口径 | 含 in-flight？ |
|---|---|---|
| JMX kafka.consumer:type=consumer-coordinates → records-lag-max | LEO − committed | ❌ |
| kafka-exporter kafka_consumergroup_lag | 同上 | ❌ |
| Confluent CC "Consumer Lag" | 同上 | ❌ |
| Spring Kafka KafkaConsumerMetrics | 同上 | ❌ |
能拿到 in-flight 的只有消费循环内部。 好消息是 Spring Kafka 提供两个无侵入扩展点：
ConsumerInterceptor（onConsume/onCommit回调）ConsumerRebalanceListener（onPartitionsAssigned/onPartitionsRevoked）
二、Per-partition in-flight gauge（推荐方案）
public class InFlightInterceptor implements ConsumerInterceptor<String, String> {
private final ConcurrentHashMap<TopicPartition, LongAdder> inFlight = new ConcurrentHashMap<>();
@Override
public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
for (ConsumerRecord<String, String> r : records) {
TopicPartition tp = new TopicPartition(r.topic(), r.partition());
inFlight.computeIfAbsent(tp, k -> new LongAdder()).increment();
}
return records;
}
@Override
public CompletionStage<OffsetsForTimestampResult> onCommit(CompletionStage<OffsetsCommitResult> future) {
future.whenComplete((result, ex) -> {
if (ex == null) {
result.offsets().forEach(tp ->
inFlight.computeIfPresent(tp, (k, v) -> { v.reset(); return v; }));
}
// ex != null → commit 失败，in-flight 保留（下次重试会覆盖）
});
return future;
}
}
暴露 Prometheus：
@Scheduled(fixedRate = 5000)
public void export() {
inFlight.forEach((tp, counter) ->
inFlightGauge.labels(tp.topic(), String.valueOf(tp.partition()))
.set(counter.sum()));
}
多分区 gap 计算：不能取 max(offset) - commitOffset（你原来的思路），因为 commit 是 per-partition 的。正确做法是 每个 partition 独立追踪：inFlight[tp] = 该 partition 已 poll 条数 − 该 partition 已 commit 条数。上面对象结构天然满足这一点。
线程卡住导致 gauge 一直非零：这恰恰是你想监控的信号，不是 bug。加一条告警：
- alert: KafkaInFlightStuck
expr: kafka_consumer_inflight_messages > 0
AND (time() - kafka_consumer_last_commit_timestamp_seconds)
> (max_poll_interval_ms / 1000) * 0.5
for: 1m
labels:
severity: warning
即：in-flight > 0 且 超过 max.poll.interval.ms 的 50% 没有 commit → 消费循环卡住了，比 rebalance 触发早暴露。
三、max.poll.records 的推导（别拍数）
max_poll_records ≤ (max_poll_interval_ms / 1000) × throughput_per_partition
你的数据：
- 日均 4500 万 / 32 分区 / 86400s ≈ 每分区每秒 ~164 条（生产速率）
- 下游处理 P99 ≈ 120ms → 单线程消费速率 ≈ 8.3 条/s
max.poll.interval.ms = 300000（5min）
理论上限 = 300s × 8.3 ≈ 2490 条。你设 500 → 最坏处理时间 = 500 × 0.12s = 60s，远小于 300s interval，安全余量充足。
但 OOMKilled 场景下，这 60s 的 500 条全部重放。建议判据：
max_poll_records × P99_single_msg_time ≤ 你能容忍的最坏重复窗口
- 如果业务幂等能兜住（Redis / DB 唯一键）→ 500 可以接受，重复代价 ≈ 0
- 如果不能（外部 RPC 不可撤回）→ 降到 200（最坏重放 200 × 0.12 = 24s，重复代价可控）
- 关键：这个值不是越大吞吐越高，它只影响「单次 poll 的批量大小」，不影响总吞吐（总吞吐 = 消费速率 × 实例数）
四、告警阈值定义
in_flight_alert_threshold = max_poll_records （结构性上限：超过说明消费循环异常）
stuck_alert_threshold = max_poll_interval_ms × 0.5 （功能性上限：超过说明卡死）
两者取 min 作为告警触发条件。阈值跟 max.poll.records 挂钩（结构性）+ 跟业务处理时间挂钩（功能性），不拍「100 条」或「30 秒」。
五、一个补充视角：rebalance 才是最大的重复源
你提到「每次 rebalance 都是一次潜在重复」。从线上数据看，rebalance 触发的重复量远大于单次 OOMKilled：
- 单次 OOMKilled：最多重复
max_poll_records × 分区数条 - 一次 rebalance（8 实例）：最多重复
8 × max_poll_records × 分区数条（所有实例的 in-flight 全部重放）
所以 max.poll.records 的优化收益在 rebalance 场景下被放大了 8 倍。如果你的消费组经常 rebalance（比如 K8s 滚动发布），降 max.poll.records 的优先级比加幂等还高。
一句话总结：lag 监控是「水位计」，in-flight gauge 是「流量计」——水位正常不代表没有漏水，你得在出水口（commit）和进水口（poll）之间装一个流量计，才能看见 OOMKilled 那 200 条的「暗流」。