Spring Kafka `acknowledge()` 与 `@Transactional` 的时序漏洞：12ms Full GC 窗口内 41 笔订单状态回滚的根因与修复

作者：江星河-885｜15 年全栈｜分布式一致性与消息中间件方向 栈：Kafka 3.5.1（KRaft，3 broker）、Spring Kafka 3.1.5、Spring Boot 3.2.5、MySQL 8.0.36、Java 17 结论先行：Spring Kafka 里 @KafkaListener 方法上加 @Transactional，Acknowledgment.acknowledge() 的默认执行时机是「业务方法 return、事务尚未 commit」，中间有一个 5–22 ms 的崩溃窗口——Full GC、broker 网络抖动、JVM SIGKILL 都会落进这个窗口，造成「Kafka 认为已消费、DB 事务回滚」的业务语义丢失。本文给可复现的时序拆解、afterCommit 修复方案与三条监控指标。

一、现象与背景：一次 12.3ms 的 Full GC，41 笔订单状态跳回 CREATED

我们的订单事件流：topic order-event，32 分区，RF=3，min.insync.replicas=2，消费组 8 实例（每实例 8C16G），日均 4200 万条，峰值 4.2 万 msg/s。业务侧用 @Transactional 包裹了「写订单状态 + 写幂等表 + 发下游 MQ」三步，Acknowledgment.acknowledge() 写在方法末尾。

去年 12 月 3 日 09:41:22，一次 12.3 ms 的 Full GC（Metaspace 扩容触发，-XX:MaxMetaspaceSize=256m 边界）让 3 个消费实例同时停顿 12.3 ms 后被 OOM-killer 杀掉（JVM 在 commit 之前收到 SIGKILL）。GC 恢复后 2 分钟内，幂等表命中率从常态 3.1% 飙到 34.7%，下游对账报「订单状态跳变」41 笔——但 Kafka 侧 lag 全程为 0，broker 日志零异常，__consumer_offsets 的写入时间线完全连续。

事后排查：这 41 笔订单的 order_status 都从 PAID 回退到了 CREATED。根因是 Full GC 停顿恰好落在「acknowledge() 已返回、@Transactional 尚未 commit」的窗口内——Kafka 已经 commit 了 offset，DB 事务还没落盘，JVM 在 commit 之前被杀掉。重启后 Kafka 从「已 commit 的 offset」继续，DB 事务回滚，下游看到的还是 CREATED。

注意：这不是「重复消费」，是更隐蔽的业务语义丢失——Kafka 不会重发（offset 已越过），DB 也没有对应记录（事务回滚了），两个存储系统各自「正确」，合起来就是丢了一笔。

二、原理：两个异步 commit 的时序

Spring Kafka 3.1.5 里，一条消息在 @KafkaListener + @Transactional 场景下的执行顺序是：

1. ConsumerRecord 从 poll() 拿到；
2. 进入 KafkaMessageListenerContainer 的 invokeListener；
3. Spring AOP 代理拦截 @Transactional，开启本地事务（TransactionSynchronizationManager 注册）；
4. 业务方法执行：写库 → 写幂等表 → 发下游 MQ；
5. 业务方法 return；
6. Acknowledgment.acknowledge() 被调用——此时 commitSync(offset) 已发出，broker 已 ACK，__consumer_offsets 已更新；
7. @Transactional 的 afterReturning 才触发 TransactionManager.commit()，向 MySQL 发 COMMIT；
8. MySQL ACK 后事务真正持久化。

步骤 6 在步骤 7 之前。 Acknowledgment.acknowledge() 的默认实现（DefaultAcknowledgment）是立即 commitSync，它不知道、也不等待 Spring 事务。所以步骤 6→7 之间有一个窗口，窗口大小 = 业务方法 return 到事务 commit 的耗时，实测 P50 8ms、P99 22ms，Full GC 期间可达 100ms+。

这个窗口的语义：Kafka 已经认为消息消费成功，DB 还没持久化。崩溃落在窗口内，就是「offset 已越过、事务回滚」——Kafka 不会重发，DB 也没有对应记录。

我旧文里写过「Kafka 的 commit 不保证『已处理』，它只保证『已收到 commit 请求』」，这篇就是把这句话落到 Spring 事务的具体时序上。

三、具体做法

3.1 反例（我们最初的写法）

@KafkaListener(topics = "order-event")
@Transactional
public void onMessage(ConsumerRecord<String, OrderEvent> record,
                      Acknowledgment ack) {
    orderRepo.updateStatus(record.value());
    idempotentRepo.mark(record.value().getBizKey());
    downstream.send(record.value());
    ack.acknowledge(); // ❌ 在事务提交前就 commit offset
}

3.2 正例 A：把 ack 挪到 afterCommit

@KafkaListener(topics = "order-event")
public void onMessage(ConsumerRecord<String, OrderEvent> record,
                      Acknowledgment ack) {
    transactionTemplate.executeWithoutResult(status -> {
        orderRepo.updateStatus(record.value());
        idempotentRepo.mark(record.value().getBizKey());
        downstream.send(record.value());
    });
    // 事务已提交（executeWithoutResult 返回即代表 commit 完成）
    ack.acknowledge(); // ✅ 在事务提交后才 commit offset
}

3.3 正例 B：用 TransactionSynchronization（方法上保留 @Transactional）

@KafkaListener(topics = "order-event")
@Transactional(rollbackFor = Exception.class)
public void onMessage(ConsumerRecord<String, OrderEvent> record,
                      Acknowledgment ack) {
    orderRepo.updateStatus(record.value());
    idempotentRepo.mark(record.value().getBizKey());
    downstream.send(record.value());
    // 不要在方法内直接 ack，改在事务提交回调里
    TransactionSynchronizationManager.registerSynchronization(
        new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                ack.acknowledge();
            }
        });
}

3.4 异常路径：不 ack，让 Kafka 重试

@KafkaListener(topics = "order-event")
@Transactional(rollbackFor = Exception.class)
public void onMessage(ConsumerRecord<String, OrderEvent> record,
                      Acknowledgment ack) {
    try {
        orderRepo.updateStatus(record.value());
        idempotentRepo.mark(record.value().getBizKey());
        downstream.send(record.value());
    } catch (Exception e) {
        throw e; // 让 @Transactional 回滚；ack 不被调用
    }
    TransactionSynchronizationManager.registerSynchronization(
        new TransactionSynchronization() {
            @Override
            public void afterCommit() { ack.acknowledge(); }
        });
}

关键点：ack 只在 afterCommit 里触发；任何异常都会让 @Transactional 回滚，afterCommit 不会被回调，Kafka 按 BackOff 配置重试。

3.5 参数基线

spring.kafka.listener.ack-mode=manual
spring.kafka.listener.concurrency=8
spring.kafka.listener.batch-listener=false
# 事务
spring.datasource.hikari.connection-timeout=3000
spring.datasource.hikari.transaction-isolation=READ_COMMITTED
# 重试
spring.kafka.listener.error-handler.backoff.fixed-delay=1000
spring.kafka.listener.error-handler.backoff.max-attempts=3

3.6 监控三件套

- kafka.consumer.commit.sync.duration   # P99 < 20ms 才算健康
- kafka.transaction.commit.duration     # 事务 commit 耗时 P99
- jvm.gc.pause.max                      # Full GC 停顿，>100ms 告警

三者叠加就是「崩溃窗口」的上界。

四、踩过的坑

1. @KafkaListener 方法内部自调用 @Transactional 不生效：Spring AOP 基于代理，同类内部调用绕过代理，事务根本没开。我们把业务逻辑抽到独立 @Service 后事务才真正生效——但 ack 的时序问题依然存在，必须配合 afterCommit。
2. finally 块里 ack.acknowledge() 是致命错误：早期我们在 finally 里兜底 ack，结果业务抛异常时事务回滚了、offset 却提交了——比不加 finally 更糟。finally 里不 ack，是这条链路的第一铁律。
3. acknowledge() 是同步阻塞的：commitSync 的 RTT 实测 P99 18ms，叠加事务 commit 的 8ms，单条处理时间从 12ms 涨到 38ms，8 实例 4.2 万 QPS 直接掉到 2.9 万。改成 commitAsync + afterCommit 回调后恢复。
4. @Transactional(readOnly=true) 也会被 Spring 开事务：即使业务只读，只要方法上有 @Transactional，就会走一次 BEGIN/COMMIT。对只读路径去掉注解，单实例吞吐提升 11%。
5. TransactionSynchronization.afterCommit 里抛异常会被吞：afterCommit 回调里如果 ack.acknowledge() 抛异常（比如 broker 不可用），Spring 会 catch 并打 warn 日志，但事务已经 commit 了——此时 Kafka offset 没提交，下次消费会重复。所以 afterCommit 里的 ack 失败必须打 metrics + 告警，不能只打日志。

五、结论与可复用清单

• ack 与 @Transactional 的时序必须显式管理：默认「方法内 ack」在崩溃窗口内会造成业务语义丢失（不是重复，是回滚后状态倒退）。
• 推荐模式：TransactionSynchronization.afterCommit 里调 ack.acknowledge()；异常路径不 ack，让 Kafka 重试。
• 不要依赖「Kafka 不重复」：即使时序对了，at-least-once 仍是默认语义，幂等表 + unique key 是兜底。
• 崩溃窗口 = commitSync RTT + 事务 commit 耗时 + GC 停顿：三者叠加，监控三件套缺一不可。
• 参数基线（32 分区、4.2 万 QPS、8 实例）：max.poll.records=200、max.poll.interval.ms=600000、session.timeout.ms=10000、heartbeat.interval.ms=3000。

一句话总结：Kafka 的 offset commit 和 DB 的 transaction commit 是两件独立的事，谁先谁后决定崩溃时是「重复」还是「丢失」——默认顺序在崩溃时是「丢失」，必须显式改成「先 DB 后 Kafka」。