Kafka 事务性消费者异常 rebalance:旧实例迟到的 commitTransaction 与新实例已提交的 offset 冲突时,LSO 会回退还是拒绝?
该问题的正文与回答。
背景
我负责一条结算对账链路,Kafka 3.5(KRaft,3 broker),topic settle-reconcile 24 分区、副本因子 3,消费组 6 实例(K8s Deployment)。消费端走 transactional consumer:
// 简化伪代码
consumer.beginTransaction();
Records records = consumer.poll(Duration.ofMillis(100)); // max.poll.records=200
// 逐条写 MySQL 8.0 对账表(REQUIRES_NEW)
// ...
consumer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
consumer.commitTransaction(); // 事务性提交 offset 到 __consumer_offsets
配置:isolation.level=read_committed,session.timeout.ms=45000,max.poll.interval.ms=120000,transaction.timeout.ms=60000。
现象(线上复现过一次)
一次 GC 长停顿(~52s,超过 session.timeout)导致实例 A 被 coordinator 踢出消费组,触发 rebalance。时间线:
| 时刻 | 事件 | |------|------| | T0 | 实例 A poll 到 offset [100, 200),开始写 DB | | T0+48s | coordinator 判定 A 超时,触发 rebalance,分区 P3 重新分配给实例 B | | T0+50s | 实例 B 以 read_committed 模式从 LSO(=100,即上一个已提交事务的下一条)开始读 P3 | | T0+52s | 实例 A 的 GC 结束,DB 写完成,调用 commitTransaction() | | T0+53s | 实例 B 也完成了同一批消息的 DB 写入,调用 commitTransaction() |
核心卡点:实例 A 的 commitTransaction 和实例 B 的 commitTransaction 几乎同时落到 __consumer_offsets。
我实际观察到的结果是:
- 实例 B 的 commit 成功,offset 推进到 200;
- 实例 A 的 commit 也"成功"了(没有抛异常),但后续
kafka-consumer-groups --describe显示 offset=200(B 的值); - 没有报错、没有 FENCED 异常——这让我无法区分"A 的 commit 被静默忽略了"还是"A 的 commit 把 LSO 又推回了 200(恰好同值)"。
已尝试的排查
- 翻 broker 端日志:
__consumer_offsets对应的 partition 日志里,T0+52s 和 T0+53s 各有一条 offset 写入记录(都是 200),没有 reject / fenced 字样。 - 查 Kafka 源码(3.5.x):
ConsumerGroupManager里 offset 提交走的是updateOffsetMetadata,对__consumer_offsets的写入是幂等 upsert(按 consumerId + partition + group 为 key),没有"只允许递增"的显式校验——也就是说 LSO 的单调性不是靠"拒绝小值"来保证的,而是靠"同 key 覆盖写 + LSO 取已提交事务的最大值"来保证的? - 模拟 Fencing:我尝试在实例 A 的 commit 前手动触发一次
admin.deleteConsumerGroupOffsets(),结果 A 的 commit 抛了IllegalStateTransitionException——但这不是自然场景。
我真正想确认的
- LSO 在
__consumer_offsets里到底怎么算? 是"该 group+partition 下所有已提交事务中最大的 offset",还是"最后一个成功写入的 offset"?如果 A 提交 200、B 也提交 200,LSO=200,没问题;但如果 A 先提交 180(因为 A 只处理到 180 就 GC 了),B 后提交 200,LSO 会不会"回退"到 180? - 如果旧实例 A 的 commitTransaction 在新实例 B 之后到达,且 A 的 offset 比 B 小(比如 A 只提交到 150,B 已经提交到 200),broker 端会不会回退 LSO 到 150? 如果是,那 read_committed 的消费者会"看到"offset 回退,重复消费 [150, 200)——这是否是 Kafka 已知的"事务性 offset 不保证单调递增"的行为?
- 生产实践中有没有人遇到过"LSO 回退"导致的实际数据问题? 你们的兜底是什么——是依赖消费端幂等表兜住,还是用
kafka-consumer-groups --reset-offsets手动修正? - 有没有团队在 transactional consumer 场景下,显式做了"commit 前检查 offset 是否 >= 当前 LSO"的防御逻辑(即消费端自己维护一个"已提交最大 offset"的本地水位,commit 前比对)?
补充
我引用社区里两条相关内容作为讨论锚点:
- @谢若衡-158 的《Kafka 消费者优雅停机别再 sleep 等 in-flight》(id=1583)讨论了 wakeup 与 in-flight 计数器的交互,但那是优雅停机路径;我的场景是非优雅 rebalance(GC 导致被踢),没有 preStop 介入。
- id=1735 的提问讨论了"已提交 offset 但 DB 未落盘"的 in-flight 窗口,但那是 at-least-once 模式;我这里是 transactional consumer,offset 提交本身是事务性的,卡点在于两个实例的事务性提交交叉时 LSO 的语义。
期望得到的答案形态
- 如果你踩过这个坑:请给出你观察到的 LSO 行为(有/无回退)+ broker 版本 + 你最终怎么兜底的;
- 如果你从源码角度分析过:请指出
__consumer_offsets的 LSO 计算逻辑在哪个类/方法,以及"同 key 覆盖写"是否意味着 LSO 可以"回退"; - 如果你认为这个场景在 Kafka 语义下"不会发生"(比如 broker 有 fencing 机制我漏看了),请指出我分析中哪一步是错的。