如何在kafka streams上实现分组转换

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KGroupedStream;
import org.apache.kafka.streams.kstream.Produced;
import java.util.Properties;

public class KafkaStreamsExample {
    public static void main(String[] args) {
        // 设置Kafka Streams应用程序的配置参数
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-streams-example");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 创建一个流构建器
        StreamsBuilder builder = new StreamsBuilder();

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

        // 将输入流按照键进行分组
        KGroupedStream<String, String> groupedStream = inputStream.groupByKey();

        // 对每个分组进行转换
        KStream<String, String> transformedStream = groupedStream.mapValues(value -> value.toUpperCase());

        // 将转换后的结果发送到输出流
        transformedStream.to("output-topic", Produced.with(Serdes.String(), Serdes.String()));

        // 创建Kafka Streams应用程序
        KafkaStreams streams = new KafkaStreams(builder.build(), props);

        // 启动应用程序
        streams.start();

        // 添加关闭钩子，确保应用程序在退出之前正常关闭
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}