如何用Embedded Kafka测试Kafka Stream处理器？为何仅@KafkaListener能收消息

当然可行！用Embedded Kafka测试Kafka Streams处理器是完全没问题的，你遇到的问题大概率是测试配置或者Streams初始化的细节没做好，我来帮你一步步排查解决：

1. 确保Embedded Kafka与Kafka Streams的配置完全对齐 #

这是最常见的坑：Streams的bootstrap.servers必须指向Embedded Kafka的实际地址，不能硬编码成默认的localhost:9092——因为Embedded Kafka默认会使用随机端口启动。同时要保证测试时的application.id唯一，避免和其他测试实例的状态存储冲突。

示例配置代码：

@Bean
public KafkaStreams kafkaStreams(EmbeddedKafkaBroker embeddedKafka) {
    Properties props = new Properties();
    // 用随机后缀保证application.id唯一
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-stream-app-" + UUID.randomUUID());
    // 动态获取Embedded Kafka的Broker地址
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    // 为测试指定临时状态存储目录，避免污染本地环境
    props.put(StreamsConfig.STATE_DIR_CONFIG, Files.createTempDirectory("kafka-streams-test").toAbsolutePath().toString());

    StreamsBuilder builder = new StreamsBuilder();
    // 这里替换成你的实际处理器拓扑
    builder.stream("input-topic")
           .mapValues(String::toUpperCase)
           .to("output-topic");

    KafkaStreams streams = new KafkaStreams(builder.build(), props);
    streams.start();
    // 注册关闭钩子，测试结束后自动清理资源
    Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    return streams;
}

2. 显式创建测试所需的主题 #

Embedded Kafka默认不会自动创建主题（除非你开启了auto.create.topics.enable=true，但测试场景下显式创建更可靠），所以在测试开始前一定要创建输入和输出主题：

@BeforeEach
void setUp(EmbeddedKafkaBroker embeddedKafka) {
    embeddedKafka.addTopics("input-topic", "output-topic");
}

3. 等待Kafka Streams完成初始化再发送消息 #

Streams启动后需要时间连接集群、初始化拓扑，直接发送消息会导致处理器“错过”消息。可以写一个工具方法，等待Streams进入RUNNING状态：

private void waitForStreamRunning(KafkaStreams streams) {
    streams.setStateListener((newState, oldState) -> {
        if (newState == KafkaStreams.State.RUNNING) {
            synchronized (streams) {
                streams.notify();
            }
        }
    });
    synchronized (streams) {
        try {
            // 最多等待30秒，避免无限阻塞
            streams.wait(30000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

在测试方法中调用这个方法，确保Streams准备就绪后再发送测试消息。

4. 正确配置测试用的生产者和消费者 #

生产者和消费者的bootstrap.servers同样要指向Embedded Kafka的地址，同时消费者的group.id要唯一，auto.offset.reset设为earliest，确保能读到所有测试消息：

@Test
void testStreamProcessing(EmbeddedKafkaBroker embeddedKafka) {
    // 1. 启动Streams并等待就绪
    KafkaStreams streams = kafkaStreams(embeddedKafka);
    waitForStreamRunning(streams);

    // 2. 发送测试消息到输入主题
    Producer<String, String> producer = createProducer(embeddedKafka);
    producer.send(new ProducerRecord<>("input-topic", "key1", "hello world"));
    producer.flush();

    // 3. 从输出主题消费并验证结果
    Consumer<String, String> consumer = createConsumer(embeddedKafka);
    consumer.subscribe(Collections.singletonList("output-topic"));
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
    
    // 这里替换成你的断言逻辑
    assertThat(records.count()).isEqualTo(1);
    assertThat(records.iterator().next().value()).isEqualTo("HELLO WORLD");

    // 4. 清理资源
    consumer.close();
    producer.close();
    streams.close();
}

private Producer<String, String> createProducer(EmbeddedKafkaBroker embeddedKafka) {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return new KafkaProducer<>(props);
}

private Consumer<String, String> createConsumer(EmbeddedKafkaBroker embeddedKafka) {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-" + UUID.randomUUID());
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return new KafkaConsumer<>(props);
}

5. 查看日志定位隐藏问题 #

如果以上步骤都做了还是不行，开启Kafka Streams的DEBUG日志，检查是否有连接失败、主题不存在、序列化/反序列化错误等信息，这些日志能帮你快速定位问题。

内容的提问来源于stack exchange，提问作者Kamil Zieliński