【读到了鬼消息：一次Kafka事务开启了但消费者没read_committed的翻车记】

【今天下午正准备摸鱼测试突然在群里拍我“订单状态乱跳先成功又回滚但消费者明明说处理过了”我一看监控Kafka消费量正常对账却有缺口脑壳嗡的一声——多半是读到了不该看的‘鬼消息’。}事故现场现象下游“订单状态同步”消费者处理了几条后来被撤销的事件导致库里出现“已支付-又被覆盖成未支付”的离谱状态同一时间段内生产者日志里有事务相关异常但消费者没报错还挺勤奋地消费了一堆。日志随手一截2026-03-21 14:02:31 INFO OrderProducer - begin txn for order1008611 2026-03-21 14:02:33 ERROR OrderProducer - txn aborted org.apache.kafka.common.errors.TransactionAbortedException: Transaction aborted消费者侧却在嘀咕2026-03-21 14:02:34 INFO OrderSyncListener - handled event {orderId1008611, statusPAID}这就诡异了生产者把事务_abort_了消费者却把消息当真的处理了。排查脑回路先看生产者这次我们为了“精准一次”EOS给生产者开了事务transactional.id链路里有多条topic需要“要么都成功要么都撤销”。生产阶段某个数据库校验失败触发了TransactionAbortedException按理整笔事务内的消息都应被过滤掉不该被消费者看见。然后我翻了消费者配置当场裂开消费者默认isolation.levelread_uncommitted也就是“读未提交”。这玩意会把失败事务里产生的record也投递给你下游还一本正经落库直接制造数据屎山。侧证把同分区的相同事件抓出来对比生产端那条对应的producerId在事务状态日志里是aborted但消费者依然收到了。板上钉钉开了生产事务却忘了把消费者设成read_committed。问题代码反例生产端Spring Kafka开了事务但配置不完整Bean public ProducerFactoryString, String pf() { MapString, Object props new HashMap(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka:9092); props.put(ProducerConfig.ACKS_CONFIG, all); props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, order-tx-001); return new DefaultKafkaProducerFactory(props); } Bean public KafkaTemplateString, String kt(ProducerFactoryString, String pf) { KafkaTemplateString, String kt new KafkaTemplate(pf); kt.setTransactionIdPrefix(order-tx-); return kt; } // 业务里用事务发多条消息 public void publishOrderEvents(Order order) { kafkaTemplate.executeInTransaction(ops - { ops.send(order-status, key(order), toJson(order)); ops.send(stock-reserve, key(order), toJson(order)); // 期间抛异常 - 整个事务 abort validateOrThrow(order); return true; }); }消费者反例默认读未提交spring: kafka: consumer: group-id: order-sync enable-auto-commit: false # 没设置isolation-level默认read_uncommitted坑ListenerKafkaListener(topics order-status, groupId order-sync) public void onMsg(ConsumerRecordString, String r, Acknowledgment ack) { OrderEvent e fromJson(r.value()); syncService.apply(e); // 这里把“已被abort的消息”也处理了 ack.acknowledge(); }根因生产者开启了事务EOS路径但消费者没设置read_committed导致读到了被abort的事务内消息一些主题混用了“事务生产者”和“非事务生产者”数据边界更乱少量分区里min.insync.replicas偏低网络抖动时NotEnoughReplicas触发重试进一步诱发事务abort。解决方案一次到位1) 消费者必须read_committed 手动提交spring: kafka: consumer: enable-auto-commit: false isolation-level: read_committed # 关键只读已提交的事务消息 max-poll-interval-ms: 600000 max-poll-records: 200KafkaListener(topics order-status, groupId order-sync) public void onMsg(ConsumerRecordString, String r, Acknowledgment ack) { try { OrderEvent e fromJson(r.value()); syncService.applyWithIdempotency(e); // 幂等兜底 ack.acknowledge(); } catch (Exception ex) { // 失败交给错误处理器/重试DQL throw ex; } }2) 生产者完整配置EOS事务幂等Bean public ProducerFactoryString, String txProducerFactory() { MapString, Object props new HashMap(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka:9092); props.put(ProducerConfig.ACKS_CONFIG, all); props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); props.put(ProducerConfig.RETRIES_CONFIG, 5); props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5); // 避免乱序 props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, order-tx-${HOSTNAME}); return new DefaultKafkaProducerFactory(props); } public void publishInTxn(ListProducerRecordString, String records) { kafkaTemplate.executeInTransaction(ops - { records.forEach(ops::send); // 可选在同一事务里写本地Outbox见下一节 return true; }); }3) 主题治理与Broker侧不要在同一个topic/partition里混用事务生产者和非事务生产者Broker参数min.insync.replicas 2配合acksall防单副本写入transaction.state.log.replication.factor 3transaction.state.log.min.isr 2为关键topic开log.cleanup.policycompact或严格保留策略避免幂等键的墓碑过早清理。4) 幂等与补偿消费侧/生产侧都要有消费侧按bizId/订单号做唯一键去重重复消息也不怕ALTER TABLE t_order_sync ADD UNIQUE KEY uk_biz (biz_id);生产侧即使事务提交失败或模板外异常也要有Outbox兜底再投递隔离出站可靠性。验证配置read_committed后复现同样的事务abort消费者不再收到那些record对账恢复一致无“先成功后回滚还被处理”的鬼畜状态压测1小时消费速率稳定重试/死信量在可控范围。踩坑总结只开生产事务不改消费者自欺欺人read_committed是下游的第一道门神。EOS不等于万无一失幂等与Outbox依然要上Broker副本/ISR也要兜住尾部风险。看见“事务abort但消息被处理”第一时间检查消费者隔离级别和主题混用情况少走弯路。