如何为带Avro序列化/反序列化的KafkaStreams编写JUnit测试用例及排障

首先，你遇到的NoSuchMethodError: org.apache.kafka.test.TestUtils.tempDirectory()异常，几乎可以肯定是Kafka相关依赖版本不一致导致的。Embedded Kafka依赖的kafka-test模块版本，和你项目中Kafka Streams、Avro序列化器等依赖的Kafka核心版本不匹配，导致方法找不到——这个方法在某些Kafka版本中被重构或移除了。

接下来给你两种解决方案，其中第一种是官方更推荐的轻量级单元测试方案：

方案一：使用TopologyTestDriver（官方推荐，无集群依赖） #

Kafka Streams提供了TopologyTestDriver工具，不需要启动真实的Kafka集群（包括Embedded Kafka），直接在内存中模拟流处理逻辑，测试速度极快，非常适合单元测试。结合Avro的话，只需要做好序列化器的配置即可。

步骤示例： #

1. 添加依赖（确保和你的Kafka Streams版本一致）：

<!-- Maven示例，Gradle类似 -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams-test-utils</artifactId>
    <version>${kafka.version}</version>
    <scope>test</scope>
</dependency>

2. 编写测试代码（结合Avro）：
   假设你有一个处理Avro格式消息的KStream拓扑，测试逻辑大致如下：

// 1. 配置Kafka Streams参数，指定Avro序列化器
Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092"); // 测试用不需要真实地址
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
// 配置Avro值序列化器（这里用Confluent的AvroSerde，需要确保依赖正确）
AvroSerde<YourAvroModel> avroSerde = new AvroSerde<>();
avroSerde.configure(Collections.singletonMap(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "mock://localhost:8081"), false);
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, avroSerde.getClass());

// 2. 创建你的Topology（比如从你的应用中获取）
Topology topology = YourStreamApplication.buildTopology();

// 3. 初始化TopologyTestDriver
try (TopologyTestDriver testDriver = new TopologyTestDriver(topology, config)) {
    // 4. 创建测试用的输入/输出Topic
    TestInputTopic<String, YourAvroModel> inputTopic = testDriver.createInputTopic("input-topic", Serdes.String().serializer(), avroSerde.serializer());
    TestOutputTopic<String, YourAvroModel> outputTopic = testDriver.createOutputTopic("output-topic", Serdes.String().deserializer(), avroSerde.deserializer());

    // 5. 发送测试用的Avro消息
    YourAvroModel testData = YourAvroModel.newBuilder().setId(1).setName("test").build();
    inputTopic.pipeInput("key1", testData);

    // 6. 验证输出结果
    KeyValue<String, YourAvroModel> result = outputTopic.readKeyValue();
    assertEquals("key1", result.key);
    assertEquals(testData.getId(), result.value.getId());
}

优势： #

• 无需启动任何Kafka集群，测试速度快
• 可以精准控制输入消息，验证每一步的处理逻辑
• 避免了Embedded Kafka带来的依赖冲突和资源占用问题

方案二：修复Embedded Kafka的依赖冲突（如果一定要用） #

如果你坚持要用Embedded Kafka做集成测试，需要严格统一所有Kafka相关依赖的版本：

1. 统一Kafka版本：在你的构建文件（pom.xml/build.gradle）中，显式指定所有Kafka模块的版本，比如：

<!-- Maven中统一版本 -->
<properties>
    <kafka.version>2.8.1</kafka.version> <!-- 替换为你使用的版本 -->
</properties>

<!-- 确保kafka-test、kafka-streams、kafka-clients等依赖版本一致 -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <version>${spring-kafka.version}</version>
    <scope>test</scope>
    <exclusions>
        <exclusion>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-test</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-test</artifactId>
    <version>${kafka.version}</version>
    <scope>test</scope>
</dependency>

2. 检查依赖树：用Maven的mvn dependency:tree或者Gradle的./gradlew dependencies命令，排查是否有冲突的Kafka依赖，手动排除不一致的版本。

总结来说，TopologyTestDriver是Kafka Streams单元测试的最优选择，它能让你快速、精准地测试流处理逻辑，而不需要处理集群启动和依赖冲突的问题。如果需要做更接近生产环境的集成测试，再考虑用Embedded Kafka并解决依赖版本问题。

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