两个不同KStreams的连接值

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.JoinWindows;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.state.Stores;

public class KStreamJoinExample {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();

        KStream<String, String> stream1 = builder.stream("input-topic-1");
        KStream<String, String> stream2 = builder.stream("input-topic-2");

        KStream<String, String> joinedStream = stream1.join(
            stream2,
            (value1, value2) -> value1 + "-" + value2, // 合并值的函数
            JoinWindows.of(Duration.ofMinutes(5)), // 连接窗口
            Materialized.as("join-store") // 状态存储
        );

        joinedStream.to("output-topic");

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