如何用Spring Boot测试Kafka Topic特定消息？Kafka Streams是否适用？

一、基础测试方案：用Spring Kafka测试组件 #

这是最直接的方案，借助Spring Kafka提供的测试工具，配合嵌入式Kafka完成验证，无需依赖外部集群。

1. 依赖配置 #

确保测试依赖中包含spring-kafka-test：

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

2. 测试代码示例 #

通过@EmbeddedKafka启动嵌入式Kafka，执行Spring Batch任务后，消费Topic并检查目标消息：

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

    @Autowired
    private JobLauncherTestUtils jobLauncherTestUtils;

    @Autowired
    private KafkaConsumer<String, YourMessageModel> testConsumer;

    @Test
    void verifyTargetMessageExistsInKafka() throws Exception {
        // 执行Spring Batch任务
        JobExecution execution = jobLauncherTestUtils.launchJob();
        Assertions.assertEquals(BatchStatus.COMPLETED, execution.getStatus());

        // 订阅目标Topic并拉取消息
        testConsumer.subscribe(Collections.singleton("your-target-topic"));
        ConsumerRecords<String, YourMessageModel> records = testConsumer.poll(Duration.ofSeconds(5));

        // 检查是否存在特定消息
        boolean targetFound = records.records("your-target-topic").stream()
                .anyMatch(record -> "expected-content".equals(record.getValue().getContent()));
        
        Assertions.assertTrue(targetFound, "未在Kafka Topic中找到目标消息");
    }

    // 配置测试用消费者
    @Bean
    public KafkaConsumer<String, YourMessageModel> testConsumer() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "batch-test-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.yourcompany.model");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 确保读取最早消息
        return new KafkaConsumer<>(props);
    }
}

二、Kafka Streams的可行性分析 #

完全可行，Kafka Streams适合复杂规则的消息验证场景，比如需要对消息做过滤、聚合、关联多Topic数据时，比普通消费者更灵活。

适用场景 #

• 需要验证消息的业务逻辑（比如统计特定消息出现次数、验证消息关联关系）
• 长期运行的集成测试，需要实时监听Topic并验证消息

实现思路示例 #

构建一个Kafka Streams拓扑，过滤出目标消息并输出到临时Topic，测试时消费该临时Topic即可验证：

@Service
public class MessageValidationStreamService {

    @PostConstruct
    public void startValidationStream() {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, YourMessageModel> sourceStream = builder.stream("your-target-topic");

        // 过滤出目标消息
        KStream<String, YourMessageModel> targetMessages = sourceStream.filter(
            (key, message) -> "expected-content".equals(message.getContent())
        );

        // 将结果输出到临时验证Topic
        targetMessages.to("validation-result-topic");

        Topology topology = builder.build();
        KafkaStreams streams = new KafkaStreams(topology, getStreamConfig());
        streams.start();
    }

    private Properties getStreamConfig() {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "message-validation-stream");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class);
        return props;
    }
}

测试时，只需消费validation-result-topic，如果能拉取到消息，即证明目标消息存在。

三、关键注意事项 #

• 嵌入式Kafka：优先使用@EmbeddedKafka，保证测试环境独立，不受外部集群影响。
• 偏移量配置：测试消费者需设置auto.offset.reset=earliest，确保能读取到Spring Batch任务写入的所有消息。
• 序列化配置：如果使用JSON格式消息，需正确配置反序列化器并指定信任包，避免反序列化失败。

内容的提问来源于stack exchange，提问作者deepika .n