为Kafka Consumer编写JUnit测试：如何模拟subscription及相关数据？

针对你提供的基于Reactive Kafka的Consumer代码，下面给出两种可行的测试方案，分别适配集成测试和单元测试场景：

方案一：嵌入式Kafka集成测试（贴近真实运行场景） #

这种方式会启动嵌入式Kafka实例，模拟真实的消息生产-消费流程，能完整覆盖订阅、消息接收和处理逻辑。

步骤： #

1. 添加嵌入式Kafka依赖
   在Maven或Gradle中加入测试依赖：

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

2. 编写测试类

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.annotation.DirtiesContext;

import java.util.concurrent.TimeUnit;

import static org.awaitility.Awaitility.await;

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"test-topic"}, ports = {9092})
@DirtiesContext
public class KafkaConsumerIntegrationTest {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private YourConsumerClass consumer; // 你的Consumer类实例

    @Test
    void testConsumerReceivesMessages() {
        // 替换Consumer配置为嵌入式Kafka地址和测试Topic
        consumer.bootstrapServerConfig = "localhost:9092";
        consumer.topicIdConfig = "test-topic";
        // SSL相关配置可在测试环境禁用或使用测试证书

        // 启动Consumer
        consumer.startConsumer();

        // 发送测试消息
        String testKey = "key1";
        Object testValue = new YourTestPayload("test-content"); // 自定义消息体
        kafkaTemplate.send("test-topic", testKey, testValue);

        // 验证消息被处理（假设你有存储已处理消息的容器）
        await().atMost(5, TimeUnit.SECONDS)
               .until(() -> consumer.getProcessedMessages().contains(testValue));
    }
}

关键说明： #

• @EmbeddedKafka会自动启动嵌入式Kafka并创建指定Topic
• 测试时需让topicIdConfig匹配测试用Topic，验证订阅逻辑是否正常
• 用Awaitility等待异步消息处理完成

方案二：Mock组件单元测试（快速验证业务逻辑） #

如果只需要验证fluxWorkflow的业务处理逻辑，无需真实Kafka交互，可以MockKafkaReceiver及相关对象，直接构造测试消息流。

步骤： #

1. 重构Consumer代码（可选但推荐）
   将KafkaReceiver的创建逻辑抽成独立方法，方便Mock：

// 在你的Consumer类中新增
protected KafkaReceiver<String, Object> createKafkaReceiver(ReceiverOptions<String, Object> options) {
    return KafkaReceiver.create(options);
}

2. 编写Mock测试

import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import reactor.kafka.receiver.ReceiverRecord;
import reactor.kafka.receiver.KafkaReceiver;

import static org.mockito.Mockito.when;

@ExtendWith(MockitoExtension.class)
public class KafkaConsumerUnitTest {

    @Mock
    private KafkaReceiver<String, Object> mockKafkaReceiver;

    @InjectMocks
    private YourConsumerClass consumer; // 你的Consumer类实例

    @Test
    void testFluxWorkflowProcessing() {
        // 构造测试用ReceiverRecord
        ReceiverRecord<String, Object> testRecord = ReceiverRecord.create(
            new org.apache.kafka.clients.consumer.ConsumerRecord<>(
                "test-topic", 0, 0L, "key1", new YourTestPayload("test-content")
            ),
            () -> {} // 模拟ack回调
        );

        // 让Mock的KafkaReceiver返回构造好的消息流
        when(mockKafkaReceiver.receive()).thenReturn(Flux.just(testRecord));

        // 启动Consumer（此时使用Mock的KafkaReceiver）
        consumer.startConsumer();

        // 验证fluxWorkflow的处理结果（假设返回处理后的Flux）
        StepVerifier.create(consumer.getProcessedFlux())
                    .expectNextMatches(result -> result.equals("processed-test-content"))
                    .verifyComplete();
    }
}

关键说明： #

• 用MockitoMockKafkaReceiver，跳过真实Kafka连接和订阅逻辑
• 直接构造ReceiverRecord作为测试数据，专注验证业务处理逻辑
• 用StepVerifier校验Reactor流的处理结果

关于subscription方法的测试说明： #

• 嵌入式Kafka方案中，只需让topicIdConfig匹配测试用Topic，即可验证订阅逻辑是否正常生效
• Mock方案中无需关注订阅逻辑，直接绕过真实Kafka交互，聚焦业务处理验证

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