Spring Boot中EmbeddedKafka测试Kafka监听器事务异常问题

生产环境中，Spring Kafka的监听器容器会自动关联KafkaTransactionManager——当配置了spring.kafka.producer.transaction-id-prefix后，容器会为消费者开启事务上下文，确保消费偏移量提交和消息发送（比如重试主题）在同一个事务里。但测试时直接调用监听器的listen方法，等于绕过了监听器容器的事务管理逻辑，没有触发Spring Kafka的事务初始化流程，导致监听器内部使用的KafkaTemplate找不到当前活跃的事务，从而抛出「No transaction is in process」异常。

方案1：模拟真实消费流程（推荐） #

不要直接调用listen方法，而是通过@EmbeddedKafka提供的测试Kafka集群发送消息到目标主题，让监听器容器自动触发消费。这种方式完全模拟生产环境的调用链路，事务会被容器自动管理。

示例代码：

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"test-topic", "retry-topic"})
class KafkaListenerTransactionalTest {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private CountDownLatch countDownLatch; // 监听器中定义的 latch 用于等待消息处理完成

    @Test
    void testListenerWithTransaction() throws InterruptedException {
        // 发送测试消息到目标主题
        kafkaTemplate.send("test-topic", "test-message");

        // 等待监听器处理完成
        countDownLatch.await(5, TimeUnit.SECONDS);

        // 验证重试主题是否收到消息（如果处理失败的话）
        // 可以用 Consumer 去拉取重试主题的消息做断言
    }
}

// 监听器示例
@Component
public class TestKafkaListener {

    private final CountDownLatch countDownLatch = new CountDownLatch(1);
    private final KafkaTemplate<String, String> kafkaTemplate;

    public TestKafkaListener(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @KafkaListener(topics = "test-topic")
    public void listen(String message) {
        try {
            // 模拟业务处理失败
            throw new RuntimeException("Process failed");
        } catch (Exception e) {
            // 发送到重试主题
            kafkaTemplate.send("retry-topic", message);
            // 抛出异常让容器回滚事务，不提交偏移量
            throw new KafkaException("Send to retry topic failed", e);
        } finally {
            countDownLatch.countDown();
        }
    }

    public CountDownLatch getCountDownLatch() {
        return countDownLatch;
    }
}

方案2：手动管理事务（适合必须直接调用listen的场景） #

如果因为测试需求必须直接调用listen方法，需要手动开启事务上下文，确保listen方法在事务中执行。可以通过@Transactional注解或者编程式事务管理实现。

方式1：使用@Transactional注解 #

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"test-topic", "retry-topic"})
class KafkaListenerDirectCallTest {

    @Autowired
    private TestKafkaListener listener;

    @Autowired
    private KafkaTransactionManager<?, ?> kafkaTransactionManager;

    @Test
    @Transactional(transactionManager = "kafkaTransactionManager")
    void testListenWithTransaction() {
        // 直接调用listen方法，此时处于事务上下文中
        listener.listen("test-message");
    }
}

方式2：编程式事务管理 #

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"test-topic", "retry-topic"})
class KafkaListenerProgrammaticTxTest {

    @Autowired
    private TestKafkaListener listener;

    @Autowired
    private KafkaTransactionManager<?, ?> kafkaTransactionManager;

    @Test
    void testListenWithProgrammaticTx() {
        TransactionTemplate txTemplate = new TransactionTemplate(kafkaTransactionManager);
        txTemplate.execute(status -> {
            try {
                listener.listen("test-message");
                return null;
            } catch (Exception e) {
                status.setRollbackOnly();
                throw e;
            }
        });
    }
}

注意事项 #

• 测试类中必须确保spring.kafka.producer.transaction-id-prefix配置生效，可以在application-test.properties中添加，或者通过@TestPropertySource注解设置：

spring.kafka.producer.transaction-id-prefix=tx-

• 监听器中使用的KafkaTemplate必须是Spring Boot自动配置的实例，它会自动关联KafkaTransactionManager，无需额外配置。

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