Spring Cloud Stream结合Kafka与Postgres事务数据库不回滚问题咨询

核心问题原因 #

你当前数据库事务不生效的根本原因是@Transactional注解错误标注在了Consumer的Bean定义方法上，Spring事务切面只会代理Spring管理Bean的公共成员方法，你返回的匿名Consumer内部的执行逻辑不会被事务切面拦截，因此数据库事务全程未启动。

问题解答 #

1. Kafka监听容器事务默认行为的问题

• 你的认知正确：只要配置了Kafka binder的transaction-id-prefix，所有Kafka监听容器都会默认开启Kafka事务，消费逻辑会运行在容器启动的Kafka事务上下文中。
• 如需指定仅部分函数运行在Kafka事务中，可对不需要事务的函数对应的绑定单独关闭事务：配置spring.cloud.stream.kafka.bindings.<binding-name>.consumer.transaction-enabled=false即可。

2. 生产者侧的事务逻辑差异

• 生产者侧逻辑确实与消费侧不同：容器层面的事务拦截只会作用于消费驱动的执行链路，手动调用KafkaTemplate、StreamBridge发送消息的场景，必须主动运行在事务上下文中（要么是消费侧容器启动的Kafka事务，要么是自定义标注@Transactional启动的事务），否则不会被自动纳入事务管理。

3. 同步Kafka与数据库事务的实现方案

要实现数据库提交优先于Kafka提交，只需按以下步骤调整即可，无需使用已废弃的ChainedTransactionManager：

1. 调整事务注解位置：将消费逻辑单独抽为Spring管理的Bean组件，在实际执行的消费方法上标注@Transactional，指定使用Spring Boot自动配置的JPA事务管理器。
   示例代码如下：

@Component
public class PersistAndSplitHandler {
    private final StreamBridge bridge;
    private final AuditLogRepository repository;

    public PersistAndSplitHandler(StreamBridge bridge, AuditLogRepository repository) {
        this.bridge = bridge;
        this.repository = repository;
    }

    @Transactional(rollbackFor = Exception.class)
    public void handle(String input) {
        bridge.send("left", ("left-" + input).getBytes());
        repository.createIfNotExists(input);

        if (input.startsWith("fail")) {
            throw new RuntimeException("Simulated error");
        }

        bridge.send("right", ("right-" + input).getBytes());
    }
}

2. 修改Consumer Bean定义：直接引用上述组件的方法作为消费逻辑：

@Bean
public Consumer<String> persistAndSplit(PersistAndSplitHandler handler) {
    return handler::handle;
}

3. 事务提交顺序保证：Spring的事务同步机制会自动按事务切面的优先级决定提交顺序，数据库事务的默认优先级高于Kafka事务，因此执行正常时会先提交数据库事务，再提交Kafka事务；抛出异常时会先回滚数据库事务，再回滚Kafka事务，完全符合你的需求。

内容的提问来源于stack exchange，提问作者Tomboyo