在Spring Boot集成测试中Mock Kafka生产者/消费者

方案1：Mock Kafka生产者，捕获发送的消息 #

如果只需要验证API是否正确触发消息发送，无需消费者参与，直接Mock KafkaTemplate 并捕获发送的业务内容：

@SpringBootTest
class KafkaApiProducerTest {

    @Autowired
    private OrderProcessingService apiService; // 你的业务API服务

    @MockBean
    private KafkaTemplate<String, OrderEvent> kafkaTemplate;

    @Test
    void testApiSendsCorrectKafkaMessage() {
        // 触发API逻辑
        apiService.processNewOrder("order-123", 299.99);

        // 捕获发送的业务消息体
        ArgumentCaptor<OrderEvent> eventCaptor = ArgumentCaptor.forClass(OrderEvent.class);
        verify(kafkaTemplate, times(1)).send(eq("order-topic"), eventCaptor.capture());

        // 验证消息内容
        OrderEvent capturedEvent = eventCaptor.getValue();
        assertEquals("order-123", capturedEvent.getOrderId());
        assertEquals(299.99, capturedEvent.getAmount());
    }
}

这种方式完全跳过真实Kafka，无任何延迟，直接验证API的生产者逻辑，无需手动构建Kafka原生消息对象。

方案2：Mock消费者，验证消息处理逻辑 #

如果API依赖消费者处理后的结果，用@SpyBean监听消费者的消息接收行为：

@SpringBootTest
class KafkaApiConsumerIntegrationTest {

    @Autowired
    private OrderProcessingService apiService;

    @SpyBean
    private OrderNotificationConsumer notificationConsumer; // 你的Kafka消费者类

    @Test
    void testApiTriggersConsumerProcessing() throws InterruptedException {
        // 触发API发送消息
        apiService.processNewOrder("order-456", 199.99);

        // 捕获消费者接收到的消息，设置短超时确保异步处理完成
        ArgumentCaptor<OrderEvent> eventCaptor = ArgumentCaptor.forClass(OrderEvent.class);
        verify(notificationConsumer, timeout(1000)).handleOrderEvent(eventCaptor.capture());

        // 验证消费者处理的消息内容
        OrderEvent receivedEvent = eventCaptor.getValue();
        assertEquals("order-456", receivedEvent.getOrderId());
    }
}

如果搭配方案1的Mock生产者，可完全避免Kafka延迟，直接验证端到端的业务逻辑传递。

方案3：使用嵌入式Kafka模拟完整流程 #

如果需要验证真实的生产者→Kafka→消费者全链路，但不想依赖外部Kafka的延迟，用@EmbeddedKafka启动内存Kafka：

@SpringBootTest
@EmbeddedKafka(partitions = 1, brokerProperties = {"listeners=PLAINTEXT://localhost:9093", "port=9093"})
class EmbeddedKafkaEndToEndTest {

    @Autowired
    private OrderProcessingService apiService;

    @Autowired
    private ConsumerFactory<String, OrderEvent> consumerFactory;

    @Test
    void testEndToEndKafkaFlow() {
        // 创建测试消费者订阅目标Topic
        Consumer<String, OrderEvent> testConsumer = consumerFactory.createConsumer();
        testConsumer.subscribe(Collections.singletonList("order-topic"));

        // 触发API逻辑
        apiService.processNewOrder("order-789", 399.99);

        // 拉取消息（内存Kafka无延迟，短超时即可）
        ConsumerRecords<String, OrderEvent> records = testConsumer.poll(Duration.ofMillis(300));
        OrderEvent receivedEvent = records.iterator().next().value();

        // 验证消息内容
        assertEquals("order-789", receivedEvent.getOrderId());
        assertEquals(399.99, receivedEvent.getAmount());

        testConsumer.close();
    }
}

这种方式保留了真实的Kafka流程特性，同时无网络延迟，可直接提取业务消息，无需手动构建Kafka原生对象。

关键说明 #

• 所有方案均围绕业务 payload 展开，无需手动创建ProducerRecord或ConsumerRecord等Kafka原生对象。
• Mock方案适合单元/轻量集成测试，嵌入式Kafka适合全链路集成测试。
• 使用ArgumentCaptor时，确保捕获类型为你的业务实体类，而非Kafka的消息包装类。

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