Kafka消费者TestContainer集成测试问题：消息发送后消费者未调用

1. 验证TestContainer Kafka连接配置 #

• 确保测试中生产者、消费者的bootstrap-servers都指向TestContainer Kafka的动态地址，不要硬编码本地地址：

  @Container
  private static final KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.4.0"));
  
  @DynamicPropertySource
  static void registerKafkaProperties(DynamicPropertyRegistry registry) {
      registry.add("spring.kafka.bootstrap-servers", kafkaContainer::getBootstrapServers);
  }
  

2. 检查消费者注解与配置 #

• 确认BCDConsumer的@KafkaListener参数：
  • topic名称完全匹配（Kafka topic大小写敏感）；
  • groupId配置正确，避免与其他测试实例冲突；
  • 若指定containerFactory，检查工厂的并发、批量消费规则是否拦截消息。
• 强制消费者从最早偏移量开始消费，避免错过测试消息：

  spring.kafka.consumer.auto-offset-reset=earliest
  

3. 解决消息异步处理的同步问题 #

Kafka消息发送与消费是异步的，测试中需等待消费完成再验证：

• 用CountDownLatch在消费者中标记处理完成：

  // 消费者类内
  private final CountDownLatch consumeLatch = new CountDownLatch(1);
  
  @KafkaListener(topics = "your-topic")
  public void handleMessage(YourMessage message) {
      // 业务逻辑
      consumeLatch.countDown();
  }
  
  // 测试类内
  producerService.sendTestMessage();
  // 等待10秒确保消费完成
  assertTrue(consumeLatch.await(10, TimeUnit.SECONDS));
  

• 或用KafkaTestUtils.getSingleRecord()主动拉取消息验证。

4. 匹配序列化/反序列化配置 #

生产者与消费者的序列化规则必须一致，比如JSON格式：

# 生产者配置
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
# 消费者配置
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.trusted.packages=*

自定义序列化类时，要确保两端配置完全对应，避免消息解析失败。

5. 测试上下文与依赖检查 #

• 测试类避免同时使用@EmbeddedKafka和TestContainer Kafka，防止端口冲突；
• 确认BCDConsumer被Spring上下文扫描到，可通过@Import(BCDConsumer.class)手动导入；
• 避免用@MockBean mock消费者的核心依赖，导致真实消费逻辑未执行。

6. 日志定位问题 #

开启Kafka DEBUG日志，排查连接、消费、序列化异常：

logging.level.org.springframework.kafka=DEBUG
logging.level.org.apache.kafka=DEBUG

重点查看：消费者是否成功连接、是否有消息接收日志、是否存在topic不存在/反序列化错误等异常。

7. 确保Topic存在 #

若Kafka关闭了自动创建topic，需手动创建测试用topic：

@Autowired
private KafkaAdmin kafkaAdmin;

@BeforeEach
void setUpTopic() {
    NewTopic testTopic = TopicBuilder.name("your-topic")
            .partitions(1)
            .replicas(1)
            .build();
    kafkaAdmin.createOrModifyTopics(testTopic);
}

内容的提问来源于stack exchange，提问作者Paolino L Angeletti