Spring Boot中@EmbeddedKafka测试的消费者配置问题解决

核心问题分析 #

你遇到的No resolvable bootstrap urls given in bootstrap.servers错误，本质是手动创建的KafkaConsumer没有指定嵌入式Kafka的地址；而无法注入EmbeddedKafkaBroker则大概率是缺失测试依赖或配置不当。

修复步骤 #

1. 确保引入Spring Kafka测试依赖 #

首先确认项目的测试依赖中包含spring-kafka-test，它提供了嵌入式Kafka的相关Bean和工具类：

Maven (pom.xml)

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

Gradle (build.gradle)

testImplementation 'org.springframework.kafka:spring-kafka-test'

2. 正确配置并使用嵌入式Kafka消费者 #

修改测试类，注入EmbeddedKafkaBroker获取嵌入式Kafka地址，同时完善消费者的必要配置：

@EmbeddedKafka(topics = {"messages"}, partitions = 1)
@SpringBootTest
@AutoConfigureMockMvc
@EnableKafka
public class ControllerTest {

    @Autowired
    private MockMvc mockMvc;

    @Autowired
    private EmbeddedKafkaBroker embeddedKafkaBroker;

    @Test
    public void testControllerSentMessage() throws Exception {
        // 发送请求触发消息生产
        mockMvc.perform(MockMvcRequestBuilders.get("/api/orders/kafka"))
                .andExpect(status().isCreated());

        // 配置消费者核心属性
        Map<String, Object> consumerProps = new HashMap<>();
        // 从嵌入式Broker获取bootstrap地址
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafkaBroker.getBrokersAsString());
        // 必须指定消费者组ID，Kafka强制要求
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
        // 重置偏移量为最早，确保能获取到测试发送的消息
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        // 允许反序列化所有包的对象，避免Json反序列化权限问题
        consumerProps.put(JsonDeserializer.TRUSTED_PACKAGES, "*");

        // 使用try-with-resources自动管理消费者资源
        try (Consumer<String, Message> consumer = new KafkaConsumer<>(consumerProps)) {
            consumer.subscribe(Collections.singleton("messages"));
            // 获取单条消息，超时时间10秒（可根据实际调整）
            ConsumerRecord<String, Message> received = KafkaTestUtils.getSingleRecord(consumer, "messages", 10000);
            
            // 断言消息内容
            assertNotNull(received, "未收到目标消息");
            assertEquals("test", received.value().getText());
        }
    }
}

3. 简化消费者创建（可选） #

使用KafkaTestUtils的工具方法可以省去手动配置的麻烦，直接创建符合要求的消费者：

// 替代手动构建消费者配置
Consumer<String, Message> consumer = KafkaTestUtils.createConsumer(
        embeddedKafkaBroker.getBrokersAsString(),
        "test-consumer-group",
        "earliest",
        StringDeserializer.class,
        JsonDeserializer.class
);
// 配置Json反序列化信任包
JsonDeserializer<Message> deserializer = (JsonDeserializer<Message>) consumer.valueDeserializer();
deserializer.addTrustedPackages("*");

consumer.subscribe(Collections.singleton("messages"));
ConsumerRecord<String, Message> received = KafkaTestUtils.getSingleRecord(consumer, "messages");

关键注意事项 #

• 必须指定消费者组ID：Kafka消费者要求必须配置group.id，否则会抛出配置异常
• 偏移量重置策略：设置auto.offset.reset=earliest能确保消费者获取到测试请求发送的历史消息，避免因消费者启动晚于生产者导致漏消息
• Json反序列化权限：如果消息是JSON格式，必须配置trusted-packages，否则会因安全限制无法反序列化自定义对象

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