kstreams-simple.java

import java.util.Properties;
import com.google.gson.JsonObject;
import org.apache.kafka.common.serialization.Serdes;
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;

// ... //

Properties props = new Properties();
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, LOCATION_OF_KAFKA_BROKERS);
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "MyStreamingApp");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

StreamsBuilder builder = new StreamsBuilder();
KStream<String, JsonObject> inputStream = builder.stream(inputTopic);

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