kstreams-join.java

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.JoinWindows;
import org.apache.kafka.streams.kstream.KStream;

// ... //

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

// left join with default serializers and deserializers
KStream<String, JsonObject> joined = inputStream.leftJoin(resultStream,
    (inputValue, outputValue) -> "record=" + inputValue + ", results=" + outputValue, /* ValueJoiner */
    JoinWindows.of(TimeUnit.MINUTES.toMillis(5))
);

joined.to(outputTopic);

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