如何用TestTopologyDriver测试无输出主题的Kafka Streams Function

你的核心问题是：待测试的process函数仅返回过滤后的KStream，但未将其绑定到输出主题，导致TopologyTestDriver无法捕获输出结果。解决思路很直接——在测试代码里显式把函数返回的KStream输出到一个测试专用主题，再通过这个主题验证结果。

第一步：修正待测试代码的语法错误 #

先把原代码里的语法问题修正，否则无法运行：

• filter的lambda表达式末尾缺少闭合括号
• 如果v是String类型，不能调用getId()，这里假设你实际处理的是带getId()方法的自定义业务对象（比如User），同步调整泛型定义：

@Bean
public Function<KStream<String, User>, KStream<String, User>> process() {
    return processKStream -> processKStream
            .filter((k, v) -> v.getId() > 10); // 补上闭合括号，适配自定义对象逻辑
}

第二步：调整测试代码，绑定输出主题 #

在测试代码中，调用process()函数后，需要把返回的KStream显式输出到一个测试主题（比如test-output-topic），让拓扑生成输出节点，这样TopologyTestDriver就能捕获数据：

// 初始化Serde，自定义对象需对应实现Serde
Serde<String> stringSerde = Serdes.String();
Serde<User> userSerde = ...; // 替换为你的User对象Serde实现

StreamsBuilder streamsBuilder = new StreamsBuilder();
TestTopology testTopology = new TestTopology(); // 假设TestTopology是你的测试类，包含process()方法

// 1. 创建输入流
KStream<String, User> inputStream = streamsBuilder.stream("input-topic", Consumed.with(stringSerde, userSerde));

// 2. 调用process函数，获取过滤后的流并绑定到测试输出主题
KStream<String, User> outputStream = testTopology.process().apply(inputStream);
outputStream.to("test-output-topic", Produced.with(stringSerde, userSerde)); // 关键步骤：绑定输出主题

// 3. 初始化TopologyTestDriver
TopologyTestDriver topologyTestDriver = new TopologyTestDriver(streamsBuilder.build());

// 4. 创建输入、输出Topic对象
TestInputTopic<String, User> inputTopic = topologyTestDriver.createInputTopic(
        "input-topic", 
        stringSerde.serializer(), 
        userSerde.serializer()
);
TestOutputTopic<String, User> outputTopic = topologyTestDriver.createOutputTopic(
        "test-output-topic", 
        stringSerde.deserializer(), 
        userSerde.deserializer()
);

第三步：编写测试验证逻辑 #

现在可以发送测试数据，验证过滤规则是否生效：

// 发送测试数据：包含符合和不符合过滤条件的对象
inputTopic.pipeInput("key1", new User(5)); // 会被过滤掉
inputTopic.pipeInput("key2", new User(15)); // 会被保留

// 验证输出结果
KeyValue<String, User> result = outputTopic.readKeyValue();
assert result.key.equals("key2");
assert result.value.getId() == 15;

// 确认无额外输出
assert outputTopic.isEmpty();

关键说明 #

• 测试时绑定的输出主题是内存模拟的临时主题，不需要在Kafka集群提前创建。
• 如果你的v确实是String类型，需调整过滤逻辑（比如基于字符串内容判断），同时对应修改Serde配置。

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