更新Kafka流链中的状态，不使用EOS方式使用Kafka流

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.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.state.Stores;

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

        // 定义输入流
        KStream<String, String> inputStream = builder.stream("input-topic");

        // 定义状态存储
        builder.addStateStore(Stores.keyValueStoreBuilder(
                Stores.inMemoryKeyValueStore("my-state-store"),
                Serdes.String(),
                Serdes.String()
        ));

        // 定义 KTable 并更新状态
        KTable<String, String> stateTable = inputStream.groupByKey()
                .reduce((value1, value2) -> value1 + " " + value2,
                        Materialized.as("my-state-store"));

        // 输出结果到另一个 topic
        stateTable.toStream().to("output-topic", Produced.with(Serdes.String(), Serdes.String()));

        // 构建并启动 Kafka Streams 应用
        KafkaStreams streams = new KafkaStreams(builder.build(), getStreamsConfig());
        streams.start();
    }

    private static Properties getStreamsConfig() {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-streams-state-update");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        return props;
    }
}