Kafka Streams 3.3.1事务操作：双输出主题的‘全有或全无’保障

要实现输入事件转换后写入两个输出主题的“全有或全无”事务保障，你需要利用Kafka Streams的**精确一次处理（Exactly-Once）**能力，同时调整代码结构确保两个写入操作处于同一个事务上下文。

核心问题分析 #

你当前的代码将输入流拆分为两个独立分支分别写入主题，这两个分支的写入操作属于不同的事务上下文，可能出现一个成功、一个失败的情况。而split/branch仅会将消息发送到第一个匹配的分支，也无法满足同时写入两个主题的需求。

解决方案步骤 #

1. 开启事务性配置 #

在StreamsConfig中设置关键参数，启用精确一次处理：

Properties props = new Properties();
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "your-broker-list");
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "your-app-id");
// 启用精确一次处理v2版本（3.3.1推荐使用）
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
// 设置事务超时时间，需小于等于broker的transaction.max.timeout.ms（默认900000）
props.put(StreamsConfig.TRANSACTION_TIMEOUT_MS_CONFIG, 300000);
// 配置基础序列化/反序列化参数
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, InputObj.serdes().getClass());

2. 调整代码结构，合并写入逻辑 #

通过flatMap将单个输入事件转换为两个不同的输出对象，再通过主题选择器路由到对应输出主题，确保两个写入操作处于同一个事务：

StreamsBuilder builder = new StreamsBuilder();
KStream<String, InputObj> stream = builder.stream("input-topic", Consumed.with(Serdes.String(), InputObj.serdes()));

// 将单个输入转换为两个输出对象，合并到同一个流中
KStream<String, Object> combinedStream = stream.flatMap((key, value) -> {
    List<KeyValue<String, Object>> outputPairs = new ArrayList<>();
    // 生成第一个输出对象
    OutputObj1 obj1 = new OutputObj1(value);
    outputPairs.add(KeyValue.pair(key, obj1));
    // 生成第二个输出对象
    OutputObj2 obj2 = new OutputObj2(value);
    outputPairs.add(KeyValue.pair(key, obj2));
    return outputPairs;
});

// 根据对象类型路由到对应主题
combinedStream.to(
    (key, value, recordContext) -> {
        if (value instanceof OutputObj1) {
            return "output-topic1";
        } else if (value instanceof OutputObj2) {
            return "output-topic2";
        }
        throw new IllegalArgumentException("未知输出对象类型");
    },
    Produced.with(
        Serdes.String(),
        // 自定义通用序列化器，适配两种输出对象
        Serdes.serdeFrom(new JsonSerializer<>(), new JsonDeserializer<>(Object.class))
    )
);

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

3. 关键注意事项 #

• 确保Kafka Broker的transaction.max.timeout.ms参数值大于等于你设置的transaction.timeout.ms，否则会触发事务超时错误。
• 生产环境中，Broker的transaction.state.log.replication.factor建议设置为3，保证事务日志的高可用性。
• 所有输出主题需支持事务（Kafka 0.11+版本默认支持）。

内容的提问来源于stack exchange，提问作者Vytautas Šerėnas