Kafka Streams多输入Topic转换输出多Topic flatMapValues使用问题

• 方法选择错误：flatMapValues 适用于1条输入生成多条输出的场景，要求返回值为 Iterable 类型（如List、Set），你当前是单条输入对应单条输出的简单转换，不需要用flatMapValues，直接用mapValues即可，你的报错大概率是返回值类型不匹配导致的。
• 方法调用错误：如果你的MyTransform是普通成员方法，在静态上下文（比如main方法）中直接调用会触发编译错误，需要将MyTransform改为静态方法，或实例化类对象后再调用。
• 逻辑覆盖不全：当前代码仅处理了0这1个Topic，没有覆盖0-9共10个Topic的需求。

方案1：基础修正版（适配你当前的写法逻辑） #

首先将MyTransform改为静态方法，同时替换flatMapValues为mapValues即可实现需求：

// 首先将自定义转换方法改为static，避免静态上下文调用报错
public static String MyTransform(String initial){
    // 你的原有转换逻辑
    ....
}

public static void main(String[] args) {
    final StreamsBuilder builder = new StreamsBuilder();
    // 循环处理0-9共10个Topic，无需重复写10遍逻辑
    for (int i = 0; i < 10; i++) {
        KStream<String, String> source = builder.stream(String.valueOf(i));
        // mapValues仅修改消息value，完全匹配你的使用需求
        source.mapValues(value -> MyTransform(value))
              .to("New" + i);
    }
    final Topology topology = builder.build();
    final KafkaStreams streams = new KafkaStreams(topology, props);
    streams.start();
    // 注册JVM关闭钩子，优雅停止流任务
    Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
}

方案2：如果你确实需要使用flatMapValues的场景（比如单条输入会生成多个输出值） #

修改MyTransform的返回值为可迭代类型即可，示例如下：

// 转换方法返回List<String>，满足flatMapValues的入参要求
public static List<String> MyTransform(String initial){
    List<String> result = new ArrayList<>();
    // 你的转换逻辑，可向result添加多个输出值
    ....
    return result;
}

// 调用部分保持flatMapValues写法即可
source.flatMapValues(value -> MyTransform(value)).to("New" + i);

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