Kafka Streams单元测试报错：未指定Value Serde或默认Serde配置

错误原因 #

你遇到的ConfigException: Please specify a value serde or set one through StreamsConfig#DEFAULT_VALUE_SERDE_CLASS_CONFIG，本质是Kafka Streams无法确定如何序列化/反序列化流中的数据。要么通过全局配置指定默认Serde，要么在拓扑的每个数据处理操作中显式声明Serde——你的测试代码既没配置全局默认Serde，拓扑里也没显式指定，所以触发了这个错误。

另外那个StateDirectory警告是因为测试默认用了系统临时目录存储状态，可能被系统自动清理，后续可以通过配置解决。

解决方案 #

有两种可行的修复方式，选其中一种即可：

方式1：在测试中配置全局默认Serde #

创建测试专用的StreamsConfig，指定默认的Key和Value Serde，传给TopologyTestDriver初始化。这样整个拓扑都会使用这些默认Serde，无需在每个操作中重复指定。

修改后的测试代码：

@BeforeEach
void setUp() {
    // 构建测试用的配置
    Properties testProps = new Properties();
    testProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-merge-topology");
    testProps.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // 测试环境用dummy地址也可
    // 设置全局默认Key Serde（你的ID是String类型）
    testProps.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    // 设置全局默认Value Serde（用JsonSerde适配你的自定义对象）
    testProps.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName());
    // 可选：指定非临时目录存储状态，消除警告
    testProps.put(StreamsConfig.STATE_DIR_CONFIG, "./kafka-streams-test-state");

    // 初始化TopologyTestDriver时传入配置
    testDriver = new TopologyTestDriver(myTopology.buildTopology(), testProps);

    Serde<MyObject1> myObject1Serde = new JsonSerde<>(MyObject1.class);
    Serde<MyObject2> myObject2Serde = new JsonSerde<>(MyObject2.class);
    Serde<MyObject3> myObject3Serde = new JsonSerde<>(MyObject3.class);

    myObject1Topic = testDriver.createInputTopic(
        MyTopology.FIRST_INPUT_TOPIC,
        Serdes.String().serializer(),
        myObject1Serde.serializer()
    );

    // 其他Topic创建逻辑保持不变
}

方式2：在拓扑中显式指定Serde #

如果不想依赖全局默认，就在拓扑构建的每个关键步骤（比如读取Topic、join操作、写入输出Topic）中显式指定对应的Serde。

示例拓扑代码修改：

public Topology buildTopology() {
    StreamsBuilder builder = new StreamsBuilder();

    Serde<String> stringSerde = Serdes.String();
    Serde<MyObject1> myObject1Serde = new JsonSerde<>(MyObject1.class);
    Serde<MyObject2> myObject2Serde = new JsonSerde<>(MyObject2.class);
    Serde<MyObject3> myObject3Serde = new JsonSerde<>(MyObject3.class);

    // 读取输入Topic时指定Serde
    KStream<String, MyObject1> stream1 = builder.stream(
        FIRST_INPUT_TOPIC,
        Consumed.with(stringSerde, myObject1Serde)
    );
    KStream<String, MyObject2> stream2 = builder.stream(
        SECOND_INPUT_TOPIC,
        Consumed.with(stringSerde, myObject2Serde)
    );

    // Join操作时指定Serde
    KStream<String, MyObject3> mergedStream = stream1.join(
        stream2,
        (obj1, obj2) -> new MyObject3(obj1.getId(), obj1.getFieldA(), obj2.getFieldB()),
        JoinWindows.of(Duration.ofSeconds(30)),
        Joined.with(stringSerde, myObject1Serde, myObject2Serde)
    );

    // 写入输出Topic时指定Serde
    mergedStream.to(
        OUTPUT_TOPIC,
        Produced.with(stringSerde, myObject3Serde)
    );

    return builder.build();
}

后续测试验证 #

修复后，你可以通过testDriver.createOutputTopic()创建输出Topic的测试实例，发送测试数据到输入Topic，然后从输出Topic读取结果，验证合并逻辑是否正确。

示例验证代码片段：

@Test
void testMergeLogic() {
    // 发送测试数据到输入Topic
    myObject1Topic.pipeInput("id1", new MyObject1("id1", "value1"));
    myObject2Topic.pipeInput("id1", new MyObject2("id1", "value2"));

    // 从输出Topic读取结果
    OutputTopic<String, MyObject3> outputTopic = testDriver.createOutputTopic(
        MyTopology.OUTPUT_TOPIC,
        Serdes.String().deserializer(),
        new JsonSerde<>(MyObject3.class).deserializer()
    );

    // 验证结果
    MyObject3 mergedObj = outputTopic.readValue();
    assertEquals("id1", mergedObj.getId());
    assertEquals("value1", mergedObj.getFieldA());
    assertEquals("value2", mergedObj.getFieldB());
}

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