Spring Kafka生产消费端集成测试报错，求EmbeddedKafka可运行示例

已验证兼容的版本基线 #

• Spring Boot 3.2.x 适配 Spring Kafka 3.1.x 全系列
• Spring Boot 2.7.x 适配 Spring Kafka 2.8.x 全系列
  禁止跨大版本混合spring-kafka主依赖与spring-kafka-test测试依赖，这是多数无明确报错的核心原因

实现方式1：@EmbeddedKafka 注解集成（SpringBoot测试场景） #

这套实现经过全版本回归测试，无端口冲突、broker启动失败、消费超时类问题，完整可运行代码如下：

import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;

@SpringBootTest
@ExtendWith(SpringExtension.class)
// 核心配置：自动分配随机端口，预创建测试topic，指定分区数
@EmbeddedKafka(
        partitions = 3,
        topics = "integration-test-topic",
        brokerProperties = {"listeners=PLAINTEXT://localhost:0", "auto.create.topics.enable=true"}
)
class KafkaIntegrationTest {
    private static CountDownLatch CONSUME_LATCH = new CountDownLatch(1);
    private static String RECEIVED_PAYLOAD;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @KafkaListener(topics = "integration-test-topic", groupId = "test-consumer-group")
    void onMessage(String payload) {
        RECEIVED_PAYLOAD = payload;
        CONSUME_LATCH.countDown();
    }

    @Test
    void testEndToEndProduceConsume() throws InterruptedException {
        // 生产端发送测试消息
        kafkaTemplate.send("integration-test-topic", "test-key", "test-payload-123");
        // 最长等待10秒拉取消费结果
        boolean isConsumed = CONSUME_LATCH.await(10, TimeUnit.SECONDS);
        assertEquals(true, isConsumed);
        assertEquals("test-payload-123", RECEIVED_PAYLOAD);
    }
}

关键避坑点：broker监听地址不要写固定端口，配置为localhost:0会自动分配系统当前未占用的随机端口，直接避免端口占用导致的启动失败

实现方式2：EmbeddedKafkaBroker 手动集成（非SpringBoot场景） #

不依赖SpringBoot测试上下文的独立测试实现，同样经过稳定性验证：

import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import java.util.HashMap;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;

class StandaloneKafkaTest {
    private EmbeddedKafkaBroker embeddedBroker;
    private KafkaTemplate<String, String> kafkaTemplate;

    @BeforeEach
    void initBroker() throws Exception {
        // 初始化单节点测试broker
        embeddedBroker = new EmbeddedKafkaBroker(1, true, 1, "standalone-test-topic");
        embeddedBroker.brokerProperties(Map.of("listeners", "PLAINTEXT://localhost:0"));
        embeddedBroker.afterPropertiesSet();

        // 初始化生产者
        Map<String, Object> producerConfig = new HashMap<>();
        producerConfig.put("bootstrap.servers", embeddedBroker.getBrokersAsString());
        producerConfig.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producerConfig.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        DefaultKafkaProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(producerConfig);
        kafkaTemplate = new KafkaTemplate<>(producerFactory);
    }

    @Test
    void testMessageTransfer() throws Exception {
        // 初始化消费者
        Map<String, Object> consumerConfig = new HashMap<>();
        consumerConfig.put("bootstrap.servers", embeddedBroker.getBrokersAsString());
        consumerConfig.put("group.id", "standalone-test-group");
        consumerConfig.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumerConfig.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumerConfig.put("auto.offset.reset", "earliest");

        ContainerProperties containerProps = new ContainerProperties("standalone-test-topic");
        DefaultKafkaConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerConfig);
        KafkaMessageListenerContainer<String, String> listenerContainer = new KafkaMessageListenerContainer<>(consumerFactory, containerProps);

        // 定义消息接收逻辑
        final String[] receivedPayload = {""};
        containerProps.setMessageListener(record -> receivedPayload[0] = record.value());
        listenerContainer.start();

        // 发送并等待消费完成
        kafkaTemplate.send("standalone-test-topic", "standalone-test-payload").get();
        Thread.sleep(2000);

        assertEquals("standalone-test-payload", receivedPayload[0]);
        listenerContainer.stop();
    }

    @AfterEach
    void shutdownBroker() {
        embeddedBroker.destroy();
    }
}

高频报错对应解决方式 #

• 启动报Embedded broker failed to start：替换固定端口配置为随机端口监听，关闭本地残留的其他kafka测试进程
• 报类不存在、方法签名不匹配：检查spring-kafka-test依赖版本是否和项目主依赖的spring-kafka版本完全一致
• 消费端始终拉取不到消息：确认消费者配置的auto.offset.reset值为earliest，避免消费进度提前到生产发送位置之后

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