How to merge multiple streams with Kafka Streams

In this tutorial, we take a look at the case when you have multiple streams, but you'd like to merge them into one.
We'll use multiple streams of a music catalog to demonstrate.

Here's the first stream of rock music:

KStream<String, SongEvent> rockSongs = builder.stream(ROCK_MUSIC_INPUT, Consumed.with(stringSerde, songEventSerde));

And the stream of classical music:

 KStream<String, SongEvent> classicalSongs = builder.stream(CLASSICAL_MUSIC_INPUT, Consumed.with(stringSerde, songEventSerde));

To merge these two streams you'll use the KStream.merge operator:

KStream<String, SongEvent> allSongs = rockSongs.merge(classicalSongs);

The KStream.merge method does not guarantee any order of the merged record streams. The records maintain their ordering relative to the original source topic.

The following steps use Confluent Cloud. To run the tutorial locally with Docker, skip to the Docker instructions section at the bottom.

Prerequisites

• A Confluent Cloud account
• The Confluent CLI installed on your machine
• Apache Kafka or Confluent Platform (both include the Kafka Streams application reset tool)
• Clone the confluentinc/tutorials repository and navigate into its top-level directory:

  git clone git@github.com:confluentinc/tutorials.git
  cd tutorials

Create Confluent Cloud resources

Login to your Confluent Cloud account:

confluent login --prompt --save

Install a CLI plugin that will streamline the creation of resources in Confluent Cloud:

confluent plugin install confluent-quickstart

Run the plugin from the top-level directory of the tutorials repository to create the Confluent Cloud resources needed for this tutorial. Note that you may specify a different cloud provider (gcp or azure) or region. You can find supported regions in a given cloud provider by running confluent kafka region list --cloud <CLOUD>.

confluent quickstart \
  --environment-name kafka-streams-merging-env \
  --kafka-cluster-name kafka-streams-merging-cluster \
  --create-kafka-key \
  --kafka-java-properties-file ./merging/kstreams/src/main/resources/cloud.properties

The plugin should complete in under a minute.

Create topics

Create the input and output topics for the application:

confluent kafka topic create rock-input
confluent kafka topic create classical-input
confluent kafka topic create all-music-output

Start a console producer:

confluent kafka topic produce rock-input

Enter a few JSON-formatted rock songs:

{"artist":"Metallica", "title":"Fade To Black"}
{"artist":"Van Halen", "title":"Jump"}

Enter Ctrl+C to exit the console producer.

Similarly, start a console producer for classical songs:

confluent kafka topic produce classical-input

Enter a few JSON-formatted classical songs:

{"artist":"Wolfgang Amadeus Mozart", "title":"The Magic Flute"}
{"artist":"Johann Pachelbel", "title":"Canon"}

Enter Ctrl+C to exit the console producer.

Compile and run the application

Compile the application from the top-level tutorials repository directory:

./gradlew merging:kstreams:shadowJar

Navigate into the application's home directory:

cd merging/kstreams

Run the application, passing the Kafka client configuration file generated when you created Confluent Cloud resources:

java -cp ./build/libs/kstreams-merge-standalone.jar \
    io.confluent.developer.MergeStreams \
    ./src/main/resources/cloud.properties

Validate that you see all songs in the all-music-output topic.

confluent kafka topic consume all-music-output -b

You should see all songs:

{"artist":"Johann Pachelbel","title":"Canon"}
{"artist":"Van Halen","title":"Jump"}
{"artist":"Wolfgang Amadeus Mozart","title":"The Magic Flute"}
{"artist":"Metallica","title":"Fade To Black"}

Clean up

When you are finished, delete the kafka-streams-merging-env environment by first getting the environment ID of the form env-123456 corresponding to it:

confluent environment list

Delete the environment, including all resources created for this tutorial:

confluent environment delete <ENVIRONMENT ID>

docker compose -f ./docker/docker-compose-kafka.yml up -d

docker exec -it broker /bin/bash

kafka-topics --bootstrap-server localhost:9092 --create --topic rock-input
kafka-topics --bootstrap-server localhost:9092 --create --topic classical-input
kafka-topics --bootstrap-server localhost:9092 --create --topic all-music-output

kafka-console-producer --bootstrap-server localhost:9092 --topic rock-input

{"artist":"Metallica", "title":"Fade To Black"}
{"artist":"Van Halen", "title":"Jump"}

kafka-console-producer --bootstrap-server localhost:9092 --topic classical-input

{"artist":"Wolfgang Amadeus Mozart", "title":"The Magic Flute"}
{"artist":"Johann Pachelbel", "title":"Canon"}

./gradlew merging:kstreams:shadowJar

cd merging/kstreams

java -cp ./build/libs/kstreams-merge-standalone.jar \
    io.confluent.developer.MergeStreams \
    ./src/main/resources/local.properties

kafka-console-consumer --bootstrap-server localhost:9092 --topic all-music-output --from-beginning

{"artist":"Metallica","title":"Fade To Black"}
{"artist":"Van Halen","title":"Jump"}
{"artist":"Wolfgang Amadeus Mozart","title":"The Magic Flute"}
{"artist":"Johann Pachelbel","title":"Canon"}

docker compose -f ./docker/docker-compose-kafka.yml down