Embedded Kafka Spring测试在实例就绪前执行，无休眠收不到监听日志

你遇到的问题本质是Kafka消息监听容器的初始化是异步流程：@KafkaListener对应的监听容器启动、与Embedded Kafka Broker建立连接、订阅Topic、完成分区分配整个流程是后台异步执行的，测试方法执行速度远快于该初始化流程，消息发送完成时消费者还未就绪，自然无法收到消息，也就没有对应日志输出。

方案1：等待监听容器完成分区分配 #

使用spring-kafka-test内置的工具类等待监听器完全就绪，是最直接的解决方案。

1. 首先给你的@KafkaListener指定唯一id：

@KafkaListener(id = "myKafkaListener", topics = "my-topic")
public class MyListener {
    // 原有监听逻辑
}

2. 测试类中注入监听容器，发送消息前等待分区分配完成：

@Autowired
@Qualifier("myKafkaListener")
private KafkaMessageListenerContainer<?, ?> listenerContainer;

@Test
void testSendEvent() throws ExecutionException, InterruptedException {
    // 等待监听器完成1个分区的分配，完全就绪
    KafkaTestUtils.waitForAssignment(listenerContainer, 1);

    Producer<Integer, String> producer = configureProducer();
    ProducerRecord<Integer, String> producerRecord = new ProducerRecord<>(TOPIC, "myMessage");
    producer.send(producerRecord).get();
    producer.close();
}

方案2：调整消费者偏移量配置避免消息丢失 #

如果不想等待容器就绪，可以修改消费者的auto.offset.reset配置为earliest，这样就算消费者订阅时消息已经发送，也会从Topic最开始的偏移量开始消费，不会漏过历史消息。
你可以在测试配置文件application-test.yml中添加如下配置：

spring:
  kafka:
    consumer:
      auto-offset-reset: earliest

方案3：使用同步断言等待消费结果（符合测试最佳实践） #

不建议通过日志观察消费结果，直接通过断言判断消费逻辑执行完成，才是单元测试的标准写法：

1. 改造你的Kafka监听器，添加测试用计数门闩：

@Component
public class MyKafkaListener {
    // 消费1条消息后门闩计数减1
    public CountDownLatch consumeLatch = new CountDownLatch(1);

    @KafkaListener(topics = "my-topic")
    public void listen(String message) {
        log.info("record received");
        // 原有业务逻辑
        consumeLatch.countDown();
    }
}

2. 测试类中等待门闩计数完成：

@Autowired
private MyKafkaListener myKafkaListener;

@Test
void testSendEvent() throws ExecutionException, InterruptedException {
    Producer<Integer, String> producer = configureProducer();
    ProducerRecord<Integer, String> producerRecord = new ProducerRecord<>(TOPIC, "myMessage");
    producer.send(producerRecord).get();
    producer.close();

    // 最多等待3秒，超时则判定测试失败
    boolean consumeSuccess = myKafkaListener.consumeLatch.await(3, TimeUnit.SECONDS);
    assertTrue(consumeSuccess, "消息未被监听器消费");
    // 每次测试后重置门闩，避免多测试用例互相影响
    myKafkaListener.consumeLatch = new CountDownLatch(1);
}

如果不想修改业务代码，也可以引入Awaitility工具类轮询断言业务结果，比如判断数据库是否生成对应记录、业务埋点是否更新等。

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