结论先行
把「已确认落库的最大 offset」做成 per-partition 的持久化 watermark，监控 committed_offset - watermark 的差值，就是你要的「未确认消息量」一等公民指标。 多实例下 watermark 语义不会乱，因为它是 per-partition 的（同一时刻只有一个实例持有某个分区），rebalance 时新实例从 watermark 之后的 offset 开始补确认即可。EOS 不覆盖消费端这个场景，但「watermark + 幂等表」的组合在纯 at-least-once 下效果等价，且不需要改 Kafka 协议层。
下面按你问的 3 条路线 + 兜底成本逐条拆。
1. 水位线方案（推荐主方案）
核心思路
在你现有的幂等表旁边，加一张 per-partition 的 watermark 表（或直接在幂等表里加一列也行，但独立表更干净）：
CREATE TABLE t_partition_watermark (
partition_id INT NOT NULL,
group_id VARCHAR(128) NOT NULL,
watermark BIGINT NOT NULL, -- 已确认落库的最大 offset（含）
updated_at DATETIME(3) NOT NULL,
PRIMARY KEY (group_id, partition_id)
) ENGINE=InnoDB;
写入时机（关键：和 offset commit 解耦）
每条消息处理完成的原子序列：
① 写业务表 + 幂等表（同一个事务）→ COMMIT
② 更新 watermark：
UPDATE t_partition_watermark
SET watermark = GREATEST(watermark, ?), updated_at = NOW(3)
WHERE group_id = ? AND partition_id = ?
③ 攒够 N 条（或每 500ms）批量 commitSync offset
步骤 ② 和 ③ 之间有一个窗口——这正是你要监控的 in-flight 区间：committed_offset - watermark。如果进程在 ② 之后、③ 之前被 kill，watermark 已经推进了但 offset 没提交，rebalance 后新实例从旧 offset 重读，幂等表会挡住重复——安全方向（宁重勿丢）。反过来，如果 ③ 之后、② 之前被 kill（offset 提交了但 watermark 没更新），committed - watermark > 0，告警触发，你立刻知道有未确认消息。
多实例 / rebalance 时 watermark 语义
这是你问的核心问题之一。答案：不会乱，原因是——
- watermark 是 per-partition 的，同一时刻一个分区只有一个消费者持有（Kafka consumer group 的基本保证）。
- rebalance 时，旧实例的
onPartitionsRevoked回调里不要清 watermark（它已经持久化在 DB 里了），新实例拿到分区后，读一下 watermark，从watermark + 1到committed_offset之间做「补确认」：逐条查幂等表，已存在的跳过，不存在的补处理。 - 补确认逻辑：
void onPartitionsAssigned(List<TopicPartition> partitions) {
for (TopicPartition tp : partitions) {
long wm = watermarkDao.get(group, tp); // 已确认到哪
long committed = consumer.position(tp); // offset 提交到哪
if (committed > wm) {
// 这段区间是「offset 已提交但可能未确认落库」的 in-flight
log.warn("in-flight gap: partition={}, committed={}, watermark={}, gap={}",
tp, committed, wm, committed - wm);
// 逐条补确认：从 wm+1 到 committed-1
for (long off = wm + 1; off < committed; off++) {
if (!idempotentDao.exists(off)) {
// 幂等表里没有 → 这条消息的 DB 写事务确实没完成
// 从 Kafka 重新拉这条消息（seek 回去），重放
consumer.seek(tp, off);
break; // seek 后从头重新消费
}
}
}
}
}
监控指标
Prometheus 暴露一个 gauge：
# 未确认消息量（per partition）
kafka_inflight_unconfirmed{group="xxx", partition="3"} 187
采集方式：每次 watermark 更新后，计算 committed_offset - watermark，写 Prometheus textfile。告警规则：
# P2：in-flight 超过 100 条持续 30s（说明 DB 写跟不上消费速度或事务被卡）
- alert: KafkaInflightHigh
expr: kafka_inflight_unconfirmed > 100
for: 30s
# P1：in-flight 超过 500 条（接近 max.poll.records，说明批量 commit 前 DB 大面积未落）
- alert: KafkaInflightCritical
expr: kafka_inflight_unconfirmed > 500
for: 10s
你那次事故的 3 个分区「offset 已前进 200+ 但幂等表查不到」，这个指标会在 rebalance 前就显示 in-flight=200+，而不是等 T+1 对账才发现。
和 offset commit 的关系
watermark 不是替代 offset commit，而是在 offset commit 之前加一道「确认」闸。offset commit 仍然是「我读到了这里」，watermark 是「我处理完了这里」。两者的差值就是你之前完全盲的 in-flight 窗口。
2. 幂等表反推（辅助/校验，不推荐做主信号）
用 SELECT COUNT(*) FROM t_idem WHERE status='pending' AND created_at > NOW() - INTERVAL 60 SECOND 近似 in-flight：
精度问题：
- 幂等表里 pending 的记录可能是「正在处理中」（正常）也可能是「处理完但 status 没推进」（bug），你分不清。
- 如果业务逻辑里有些消息不走幂等表（比如纯查询类消息），这个计数会低估 in-flight。
- 每 15s 跑一次 COUNT(*) 在 400 万行/天的表上有索引压力（
(status, created_at)复合索引可以缓解，但高频扫描仍不优雅）。
建议：把它当交叉校验用——watermark 说 in-flight=0，但幂等表里还有 pending 超过 5 分钟的记录，说明有 bug（watermark 推进了但业务写没完成）。两者矛盾时告警。
3. Kafka 原生 EOS 能不能覆盖
不能直接覆盖，但思路可以借鉴。
- Kafka 的 transactional producer / EOS 解决的是生产端的「消息 + 状态变更」原子性（producer 在事务里写 Kafka + 更新状态，要么都成功要么都回滚）。
- 消费端的 in-flight 窗口是「消费者拉了消息 → 写 DB → 提交 offset」之间的间隙，Kafka 协议层不提供「消费者事务」的原子性保证。
- 但有一个间接用法：如果你把「写业务表」也改成通过 Kafka 事务来写（比如用 Kafka Connect 的 SMT 或自定义 sink），那 EOS 可以覆盖「Kafka 消息 → DB 写入」的原子性。但这等于把 DB 写入搬到了 Kafka 事务里，架构改动太大，不推荐。
更实际的做法：你现在的 at-least-once + 幂等表已经是正确的消费端模型，watermark 只是在这个模型上加一层可观测性，不需要改协议。
4. 兜底成本估算（你问的「算过账吗」）
你的场景：8 实例 × 16 分区 × 每天 2~3 次 rebalance。
| 成本项 | 估算 |
|---|---|
| watermark 表写入 | 每批 commit 前多 1 次 UPDATE（8 实例 × 16 分区 = 最多 128 行，但实际同时活跃的远不到），对 4k msg/s 的吞吐影响 < 1% |
| rebalance 补确认 | 每次 rebalance 最多补 max.poll.records 条（50 条/分区 × 16 分区 = 800 条幂等表查询），按 PK 查，< 50ms |
| 监控开销 | 1 个 Prometheus gauge，可忽略 |
| T+1 对账成本（现状） | 你那次事故补跑了 40 分钟 + 人工排查，这才是最大的隐性成本 |
watermark 方案的增量成本（多 1 张 16 行的小表 + 每次 commit 前 1 次 UPDATE）远低于「rebalance 丢消息 → T+1 对账 → 补跑 → 人工排查」的兜底成本。
5. 避坑清单（我踩过的）
- watermark 更新和 offset commit 之间不要加 sleep 或复杂逻辑——这个窗口越大，in-flight 越大。理想情况是 watermark UPDATE 和 commitSync 之间只有网络 RTT。
onPartitionsRevoked里不要删 watermark 记录——rebalance 后新实例要靠它做补确认。只有当 watermark 追平 committed offset 且分区被永久移除时才清理。- watermark 只增不减（用
GREATEST）——防止 rebalance 后旧实例的延迟写入把新实例的 watermark 拉回去。 - 补确认时 seek 要 seek 到最早的那条未确认 offset，不要逐条 seek——一次 seek 到
wm+1然后顺序重放，幂等表挡住已处理的。 - kill -9 场景：你提到的「内存值蒸发」问题，watermark 持久化在 DB 里就不存在这个问题。进程被 kill 后 watermark 停在最后一个成功 UPDATE 的位置，新实例补确认时自然覆盖。
一句话总结
offset commit 是「我读到了」，watermark 是「我处理完了」，两者的差值就是你之前完全盲的 in-flight 窗口。 加一张 16 行的小表 + 每次 commit 前多 1 次 UPDATE，就能把「T+1 对账才发现」变成「rebalance 前 10 秒就告警」。