Kafka 事务的原子性边界是 PID epoch 不是进程：双 producer 共用 transactional.id 后 1.8 万条消息静默失败，而 read_committed 反而放行了脏数据

栈：Kafka 3.5.2（KRaft，3 broker，8C32G）、Spring Kafka 3.1.5、kafka-clients 3.5.2、Java 17（G1）、MySQL 8.0.36 一句话结论：Kafka 事务的原子性边界是「transactional.id + PID epoch」，不是进程、更不是线程。一旦同一进程里有两个 producer 对象共用同一个 transactional.id，后 initTransactions 的那个会把前一个的 epoch 顶掉——前一个从此每发一条都被 ProducerFencedException 拒绝，而它负责的监控埋点 producer 一切正常，监控全绿。 更阴的是另一半：你以为都在事务里的那批消息，其实有一条是 executeInTransaction 之外发的普通 send，它没有事务 marker，read_committed 消费端根本挡不住它。

一、现象：一次发布后的「监控全绿、订单全丢」

去年 11 月我们把订单服务从「本地消息表」迁到「Kafka 事务」，目标是把「UPDATE 库存 + 发 order-created」做成原子。发布后第一个业务日（约 6 小时）出现三件互相矛盾的事：

1. 订单表 UPDATE 成功行数 18,240，但下游库存服务收到的 order-created 事件只有 611 条——差了 17,629 条；
2. producer 应用没有任何 ERROR 日志，KafkaTemplate 的 record-send metrics 全 0 失败；
3. 审计日志 topic order-audit 里却有 18,240 条「订单已创建」记录，和订单表完全对得上。

也就是说：订单表说「18,240 笔都成功了」，审计日志也这么说，但下游一条没收到。三个数据源两两矛盾，而监控大盘一片绿。我们花了整整一个下午才定位到：这不是 broker 的问题，是 producer 端两个 producer 对象在「抢」同一个 transactional.id。

二、原理：fence 是 broker 端的「防旧 PID 复活」，不是「防并发」

Kafka 事务的原子性，靠的不是两阶段提交，而是三样东西配合：PID（Producer ID）+ epoch 的围栏（fence）、abort index / LSO 的可见性裁剪、事务 marker（commit/abort）落盘。

关键机制一：epoch 递增即 fence。同一个 transactional.id，每次 initTransactions() 都会走 InitProducerId RPC，broker 给这个 id 的 epoch +1。一旦某 producer 持有的 epoch 落后于 broker 记录的当前值，它再发任何消息，broker 用错误码 102 (PRODUCER_FENCED) 拒绝，客户端（kafka-clients 2.5+）抛 ProducerFencedException（旧名 UnknownProducerFenceException）。注意：这个拒绝发生在 broker 端、发生在 send 的 RPC 阶段，但 Spring Kafka 的 KafkaTemplate 在某些重试配置下会把 fenced 异常吞进 send 的 CompletableFuture 里，如果你没 .get() 或没配 errorHandler，应用主线程根本感知不到失败——这正是「监控全绿」的来源。

关键机制二：LSO 只挡「pending 事务」，不挡普通消息。read_committed 消费端能读到哪，取决于 LSO（Last Stable Offset），而 LSO 的推进只受「尚未 commit/abort 的事务」影响。一条没有事务 marker 的普通消息，不落在任何 pending 事务区间里，所以 LSO 不会为它冻结——read_committed 对它完全放行。换句话说，你以为 read_committed 是「只读已提交」的保险，但它只保险「事务消息」，对「混进来的普通消息」它是睁一只眼闭一只眼的。

把这两点拼起来，本次事故就有完整链路：

进程启动
 ├─ Producer A（业务事务）  initTransactions()  → epoch 0 → 1   ✅ 持有 epoch=1
 └─ Producer B（监控埋点）  initTransactions()  → epoch 1 → 2   ✅ 持有 epoch=2，fence 掉 A
     （B 后启动，broker 认为「epoch=2 才是当前」）
业务线程（用 A）：executeInTransaction { UPDATE 库存; send order-created }
     → A 的 send 走 InitProducerId 后 epoch 已过期
     → broker 返回 102 PRODUCER_FENCED
     → CompletableFuture 异常，但主线程未 .get()，无日志、metrics 未记失败
     → 17,629 条 order-created 全部静默失败
审计线程（用 B）：send order-audit
     → B epoch=2 有效，普通 send 成功
     → 18,240 条审计记录全部落 broker，read_committed 放行

于是：订单表（DB 事务里 UPDATE 成功，但这里其实 DB 也回滚了一部分——见下节）+ 审计日志（B 发的，全在）+ 下游（A 发的，几乎全丢）。

三、具体做法：怎么发现、怎么修

1. 先确认「是不是 fence」——一条命令看 broker 端 epoch：

# 看 __transaction_state 里该 transactional.id 当前的 epoch
kafka-consumer-groups.sh --bootstrap-server broker1:9092 \
  --describe --group __transaction_state   # 3.5 起建议用 metadata-shell
# 更直接：broker 日志 grep
grep "PRODUCER_FENCED\|transactional.id=order-pay" broker.log
# 客户端侧：把 send 的 future 强制 .get()，让异常显形
template.send(Topic, key, value).get(2, TimeUnit.SECONDS)  # 别裸调 send

2. 根因修复——一个进程只允许一个事务 producer，且 transactional.id 必须全局唯一：

// ❌ 反例：两个 bean 都 set 同一个 transactional.id
@Bean KafkaProducer biz()   { p.put(TRANSACTIONAL_ID, "order-pay"); ... }
@Bean KafkaProducer metric(){ p.put(TRANSACTIONAL_ID, "order-pay"); ... }  // 必炸

// ✅ 正例 A：监控埋点根本不该进事务，用独立 id 或干脆不开事务
@Bean KafkaProducer metric(){ p.put(TRANSACTIONAL_ID, "order-metric"); p.put(ENABLE_IDEMPOTENCE, true); ... }

// ✅ 正例 B：监控走非事务 producer，业务走事务 producer，进程内各一个

3. 兜底——给 KafkaTemplate 配 errorHandler，杜绝「静默失败」：

@Bean
KafkaTemplate<String, OrderEvent> template(ProducerFactory<String, OrderEvent> pf) {
  KafkaTemplate<String, OrderEvent> t = new KafkaTemplate<>(pf);
  t.setTransactionId("order-pay");
  // 关键：fenced/致命异常必须显式失败，不能吞
  t.setDefaultTimeout(Duration.ofSeconds(2));
  return t;
}
// 业务侧：executeInTransaction 内的每个 send 都要 .get() 或依赖模板的异常传播
template.executeInTransaction(ops -> {
  inventoryDao.deduct(orderId);
  ops.send(Topic, orderId, event).get(2, TimeUnit.SECONDS); // 强制同步暴露 fenced
  return null;
});

4. 容量与告警——给「DB 成功数 vs 下游收到数」加对账告警：

-- 对账：订单表成功 vs audit 落库 vs 下游确认，三者差 > 0 即告警
SELECT date,
       COUNT(*) db_ok,
       (SELECT COUNT(*) FROM order_audit a WHERE a.order_id IN (SELECT id FROM orders o WHERE ...)) audit_ok
FROM orders WHERE status='CREATED' AND created_at >= NOW() - INTERVAL 1 DAY;
-- 我们事后用这条对账 SQL 兜住了 3 次同类隐患

四、踩过的坑与反直觉之处

1. read_committed 不是「只读已提交」的全能保险：它只裁剪「pending 事务」区间，对没有事务 marker 的普通消息一律放行。所以「事务里混一条非事务 send」，read_committed 消费端会把它当已提交数据直接消费——隔离级别挡不住你自己在 producer 端制造的脏数据。
2. fence 失败可以完全静默：ProducerFencedException 落在 CompletableFuture 里，裸调 send() 不 .get()、又没配 errorHandler 时，主线程无感知、record-send metrics 也不记失败。「监控全绿」恰恰是最危险的信号。
3. 两个 producer 共用 transactional.id，谁后 initTransactions 谁赢：被 fence 的那个不是「报错崩溃」，而是「每条都发不出去」，且它若同时承担监控埋点，监控反而证明「进程活着」。
4. epoch 是 broker 端 LRU 维护的：即使你只有一个 producer，长时间空闲后 PID 被 __transaction_state 滚动淘汰，重启后也会撞 fence——所以 transactional.id 不要复用得太「死」，必要时带 pod-uuid 后缀重发。
5. 事务内的 DB 操作和事务外的 DB 操作不原子：我们最初把 inventoryDao.deduct 放在了 executeInTransaction 之外「为了性能」，结果 DB 扣减成功、Kafka fenced 失败，出现了「库存扣了、订单没发」的第三类脏数据。

五、结论与可复用清单

• Kafka 事务的原子性边界 = transactional.id + PID epoch，不是进程/线程。同一 transactional.id 全进程只允许一个事务 producer。
• 监控/埋点类发送不要和事务发送共用 transactional.id；要么独立 id，要么干脆不开事务（at-least-once + 幂等）。
• executeInTransaction 内的每个 send 都要 .get() 或依赖模板异常传播，杜绝 ProducerFencedException 静默进 future。
• read_committed 只挡 pending 事务，不挡普通消息：producer 端混入的非事务 send 会直接穿透，必须靠「全在事务内」+「幂等」双兜底。
• 加一条「DB 成功数 vs 下游收到数」的对账告警：本次 17,629 条的缺口，靠对账 SQL 在 T+0 就能抓到，而不是等客服工单。

留一个开放问题：当 transactional.id 数量很大（比如按 order-{orderId} 动态生成）时，broker 端 __transaction_state 的 LRU 淘汰 + epoch 重分配会不会成为新的瓶颈？动态 id 的「短生命周期 fence」和静态 id 的「长生命周期 fence」，在 PID 分配、__transaction_state 写放大上到底差多少？目前我只有直觉、没有压测数据——动态 transactional.id 的 fence 成本，是否值得为「每单独立事务」这个语义去付？ 如果你有过大规模动态 id 的事务压测，很想听听你的数字。