Java版Kafka Streams：实现符合条件的消息多主题分发

需求：监听一个Kafka Topic，消息处理后分发至三个不同的输出主题，且同一消息可进入多个输出主题。

我尝试了以下代码：

Map<String, KStream<String, String>> branches = builder
        .stream("input", Consumed.with(Serdes.String(), Serdes.String()))
        .transform(supplier1, "TRANSIT_STORE_NAME")
        .split(Named.as("prepare_data"))
        .branch((k, v) -> v != null && v.contains("xyz"), Branched.as("xyz"))
        .branch((k, v) -> v != null && v.contains("abc"), Branched.as("abc"))
        .noDefaultBranch();
branches.get("prepare_dataxyz")
        .transform(supplier2)
        .to("output.xyz");
branches.get("prepare_dataabc")
        .transform(supplier2)
        .to("output.abc");

问题：理论上，值为abcxyz_blabla的记录应该同时进入output.xyz和output.abc主题，但branch()会将消息发送至第一个匹配的分支，没法实现多主题分发。

核心问题分析 #

split().branch()是排他性分支逻辑：一条消息只会匹配第一个满足条件的分支，之后不会再进入后续分支，天生不满足“同一消息进入多个主题”的需求。

正确实现方式 #

基于处理后的原始流，为每个输出主题创建独立的过滤分支。Kafka Streams会自动复制流数据，同一条消息可以被多个过滤器匹配，从而发送到多个目标主题。

示例代码：

// 先完成统一的前置处理，得到中间流
KStream<String, String> processedStream = builder
        .stream("input", Consumed.with(Serdes.String(), Serdes.String()))
        .transform(supplier1, "TRANSIT_STORE_NAME");

// 针对每个输出主题，单独创建过滤+处理+发送逻辑
// 发送到output.xyz
processedStream
        .filter((k, v) -> v != null && v.contains("xyz"))
        .transform(supplier2)
        .to("output.xyz");

// 发送到output.abc
processedStream
        .filter((k, v) -> v != null && v.contains("abc"))
        .transform(supplier2)
        .to("output.abc");

// 第三个输出主题（按需添加）
processedStream
        .filter((k, v) -> v != null && v.contains("def"))
        .transform(supplier2)
        .to("output.def");

逻辑说明 #

• 每个filter()操作都是基于processedStream的独立流分支，Kafka Streams会负责复制消息到各个分支，不会出现排他性拦截。
• 只要消息满足某个分支的过滤条件，就会进入对应的处理流程并发送到目标主题，同一条消息可以同时匹配多个分支。

内容的提问来源于stack exchange，提问作者Dmitriy M.