Spring Cloud Stream测试时无法连接EmbeddedKafka问题求助

针对你在测试Spring Cloud Stream应用时遇到的EmbeddedKafka收不到消息、Kafka头丢失、无@EmbeddedKafka仍能启动等问题，给出具体排查和解决步骤：

1. 未加@EmbeddedKafka却能启动的原因 #

Spring Cloud Stream在测试环境下默认启用绑定器测试支持，自动用内存消息通道（如DirectChannel）替代真实Kafka Broker，所以没启动Kafka也能正常启动应用。要强制使用嵌入式/真实Kafka，需：

• 在测试类上添加@EmbeddedKafka注解，或导入EmbeddedKafkaBrokerConfiguration
• 在application-test.yml中指定默认绑定器为kafka：

spring:
  cloud:
    stream:
      default-binder: kafka

2. EmbeddedKafka接收不到消息的排查步骤 #

依赖检查 #

确保spring-kafka-test依赖已正确引入（版本需与Spring Boot、Spring Cloud Stream匹配）：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>

测试类配置 #

测试类需同时启用@SpringBootTest、@EmbeddedKafka，并绑定正确的通道：

@SpringBootTest
@EmbeddedKafka(partitions = 1, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"})
@EnableBinding(YourChannelInterface.class)
public class ProducerValidationTest {

    @Autowired
    private MessageChannel yourOutputChannel;

    // 测试逻辑
}

消息同步问题 #

测试时必须等待消息发送完成再验证，可用CountDownLatch做同步：

@Test
public void testMessageDelivery() throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(1);

    // 初始化嵌入式Kafka消费者
    Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("test-group", "true", embeddedKafka);
    DefaultKafkaConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps);
    ContainerProperties containerProps = new ContainerProperties("your-target-topic");
    containerProps.setMessageListener((MessageListener<String, String>) message -> {
        // 验证消息内容和头信息
        latch.countDown();
    });
    KafkaMessageListenerContainer<String, String> container = new KafkaMessageListenerContainer<>(consumerFactory, containerProps);
    container.start();

    // 发送消息
    yourOutputChannel.send(MessageBuilder.withPayload("test-content").build());

    // 等待消息接收，超时5秒
    latch.await(5, TimeUnit.SECONDS);
    assert latch.getCount() == 0;
}

3. Kafka头信息丢失的解决 #

测试环境下SCS默认过滤Kafka原生头，需在配置中允许所有头传递：

spring:
  cloud:
    stream:
      kafka:
        binder:
          headers: "*"  # 允许所有头信息
          configuration:
            header.format: kafka  # 使用Kafka原生头格式

发送消息时要明确设置Kafka原生头，而非仅用SCS通用头：

Message<?> message = MessageBuilder.withPayload("test")
        .setHeader(KafkaHeaders.TOPIC, "your-topic")
        .setHeader("custom-kafka-header", "header-value")
        .build();
yourOutputChannel.send(message);

4. 生产者测试收不到消息的额外检查 #

• 确认测试topic已创建：EmbeddedKafka不会自动创建topic，需在@EmbeddedKafka中指定：

@EmbeddedKafka(topics = "your-target-topic", partitions = 1)

• 检查消费者group-id，测试环境用独立group-id避免offset冲突
• 验证序列化/反序列化配置，确保生产者和消费者用相同的序列化方式

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