如何更改@KafkaStreamsStateStore kafka Stream Cloud的默认serdes

@KafkaStreamsStateStore kafka Stream Cloud的默认serdes是指Kafka Streams应用程序在处理状态存储时使用的序列化和反序列化器。默认情况下，Kafka Streams使用的是AvroSerde，它基于Avro格式进行序列化和反序列化。

要更改@KafkaStreamsStateStore kafka Stream Cloud的默认serdes，可以按照以下步骤进行操作：

创建自定义的序列化和反序列化器：根据应用程序的需求，可以选择使用不同的序列化和反序列化器。例如，可以使用JSON、Protobuf、或自定义的序列化和反序列化器。
在Kafka Streams应用程序中配置自定义的serdes：在应用程序的配置中，指定使用自定义的序列化和反序列化器。这可以通过在应用程序的配置文件中设置相应的属性来实现。
注册自定义的serdes：在应用程序的拓扑中，使用StreamsBuilder#stream()或KStream#through()方法创建流或表时，使用.withValueSerde()和.withKeySerde()方法注册自定义的serdes。

下面是一个示例代码片段，展示了如何更改@KafkaStreamsStateStore kafka Stream Cloud的默认serdes为JSON序列化和反序列化器：

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KStreamBuilder;

import java.util.Properties;

public class KafkaStreamsApp {
    public static void main(String[] args) {
        Properties config = new Properties();
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-streams-app");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 设置自定义的序列化和反序列化器
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        KStreamBuilder builder = new KStreamBuilder();
        KStream<String, String> stream = builder.stream("input-topic");

        // 注册自定义的serdes
        stream.to("output-topic", Produced.with(Serdes.String(), Serdes.String()));

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

在上述示例中，我们使用了JSON序列化和反序列化器（Serdes.String()）作为自定义的serdes，并将其应用于输入和输出流。