Spring Boot中Kafka事务与RetryableTopic能否兼容使用？

问题 #

我有一个配置了MongoDB数据库的Spring Boot应用，使用Spring的Kafka和Mongo库。我有一个Kafka监听器，调用带有@Transactional注解的方法，该方法会写入多个MongoDB集合，最后向另一个Kafka主题发送消息。若抛出异常，Mongo事务管理器会回滚所有更改，但因未配置Kafka事务管理器，消息仍会提交。

我最近添加了@RetryableTopic注解，在抛出异常时消费者会重新执行逻辑，未配置Kafka事务时此功能正常。

但当我配置Kafka事务管理器后，重试逻辑无法正常工作，出现“transaction was marked as rollback only”异常，消息未发送到重试主题，而是陷入重复读取原主题同一消息的循环。

我查阅了Spring Kafka文档，看到提示“非阻塞重试无法与容器事务结合。当监听器代码抛出异常时，容器事务提交成功，记录会发送到重试主题。”但我未看到消息写入重试主题，不确定此提示是否适用于我的场景。

请问我遗漏了什么？或者Spring的RetryableTopic无法实现此需求？以下是监听器的伪代码：

public class Consumer {
    @KafkaListener(/* configuration */)
    @RetryableTopic(/* configuration */)
    public void consume(ConsumerRecord record) {
        service.foo(record);
    }

    // in a service class 
    @Transactional
    public void foo(ConsumerRecord record) {
        /* logic */
        mongoTemplate.save(object1);
        mongoTemplate.save(object2);
        kafkaTemplate.sendMessage(message);
    }
}

解答 #

核心问题原因 #

• 事务边界冲突：配置Kafka事务管理器后，容器会开启Kafka事务管理消费偏移提交。而@Transactional注解的方法开启的Mongo事务抛出异常时，会将事务标记为回滚，这个状态会传播到外层的Kafka容器事务，导致Kafka事务也被标记为回滚。
• 非阻塞重试与容器事务不兼容：@RetryableTopic属于非阻塞重试，它需要容器正常提交Kafka事务，才能将原消息转发到重试主题。但Kafka事务被标记回滚后，容器无法完成提交，也就无法执行转发逻辑，只能重复消费原消息。

可行解决方案 #

方案1：拆分事务边界，隔离Mongo与Kafka事务

将Mongo数据库操作和Kafka消息发送拆分为独立的逻辑，避免Mongo事务的回滚状态传播到Kafka容器事务：

public class Consumer {
    @KafkaListener(/* configuration */)
    @RetryableTopic(/* configuration */)
    public void consume(ConsumerRecord record) {
        // 先执行Mongo事务操作
        service.executeMongoOperations(record);
        // 事务提交成功后再发送Kafka消息
        service.sendKafkaMessage(record);
    }
}

@Service
public class BizService {
    // 指定使用Mongo事务管理器，避免与Kafka事务冲突
    @Transactional(transactionManager = "mongoTransactionManager")
    public void executeMongoOperations(ConsumerRecord record) {
        mongoTemplate.save(object1);
        mongoTemplate.save(object2);
    }

    public void sendKafkaMessage(ConsumerRecord record) {
        kafkaTemplate.sendMessage(message);
    }
}

这样Mongo事务的回滚只会影响数据库操作，不会干扰外层Kafka容器事务，监听器抛出异常时，容器能正常提交事务并将消息转发到重试主题。

方案2：切换为阻塞重试机制

如果业务允许在当前消费线程内重试，用@Retryable替代@RetryableTopic，配合@Recover处理最终失败场景：

public class Consumer {
    @KafkaListener(/* configuration */)
    @Retryable(value = {Exception.class}, maxAttempts = 3)
    public void consume(ConsumerRecord record) {
        service.foo(record);
    }

    // 重试耗尽后的兜底处理
    @Recover
    public void handleFailedConsume(Exception e, ConsumerRecord record) {
        // 例如：记录告警日志、转发到死信主题
    }
}

这种方式不依赖Kafka事务提交来转发重试消息，事务回滚后直接在当前线程重试，避开了事务冲突问题。

方案3：手动管理偏移与重试逻辑

关闭容器自动提交偏移，手动控制偏移提交和重试消息发送：

• 在容器配置中设置autoCommitOffset = false；
• 业务逻辑执行成功时，手动调用Acknowledgment.acknowledge()提交偏移；
• 业务异常时，手动将消息发送到重试主题，再提交原消息的偏移。

这种方式灵活性最高，但需要自行处理重试次数跟踪、死信队列等细节。

内容的提问来源于stack exchange，提问作者דוד פאר