生成一个kafka streams 调用kafka java代码

以下是一个简单的Kafka Streams应用程序，它从一个名为“input_topic”的Kafka主题中读取字符串消息，并将它们转换为大写字符串，然后将它们写入另一个名为“output_topic”的Kafka主题中。

import java.util.Properties;
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.Consumed;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;

public class MyKafkaStreamsApp {

  public static void main(String[] args) {

    // Define Kafka Streams properties
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("application.id", "my-kafka-streams-app");

    // Create a StreamsBuilder object
    StreamsBuilder builder = new StreamsBuilder();

    // Create a KStream object by reading from the input topic
    KStream<String, String> inputStream = builder.stream(
        "input_topic", 
        Consumed.with(Serdes.String(), Serdes.String())
    );

    // Transform each message to uppercase
    KStream<String, String> outputStream = inputStream.mapValues(
        value -> value.toUpperCase()
    );

    // Write the transformed messages to the output topic
    outputStream.to(
        "output_topic", 
        Produced.with(Serdes.String(), Serdes.String())
    );

    // Create a KafkaStreams object and start the application
    KafkaStreams streams = new KafkaStreams(builder.build(), props);
    streams.start();
  }
}

在这个例子中，我们首先定义了一些Kafka Streams属性，然后创建了一个StreamsBuilder对象。接着，我们使用builder对象从“input_topic”读取消息，并使用mapValues()方法将消息转换为大写字符串。最后，我们使用to()方法将转换后的消息写入“output_topic”。

最后，我们创建了一个KafkaStreams对象，并启动了应用程序。当启动应用程序时，Kafka Streams将自动创建并管理所需的Kafka主题分区。