How to dynamically choose output topics with Kafka Streams

Consider a situation where you want to direct the output of different records to different topics, like a "topic exchange." In this tutorial, you'll learn how to instruct Kafka Streams to choose the output topic at runtime, based on information in each record's header, key, or value.

  builder.stream(INPUT_TOPIC, Consumed.with(stringSerde, orderSerde))
         .mapValues(orderProcessingSimulator)
        .to(orderTopicNameExtractor, Produced.with(stringSerde, completedOrderSerde));

Here's our example topology. To dynamically route records to different topics, you'll use an instance of the TopicNameExtractor. As shown in here, you provide the TopicNameExtractor to the overloaded KStream.to.

Here's the TopicNameExtractor used in this example. It uses information from the value to determine which topic Kafka Streams should use for this record.

final TopicNameExtractor<String, CompletedOrder> orderTopicNameExtractor = (key, completedOrder, recordContext) -> {
              final String compositeId = completedOrder.id();
              final String skuPart = compositeId.substring(compositeId.indexOf('-') + 1, 5);
              final String outTopic;
              if (skuPart.equals("QUA")) {
                  outTopic = SPECIAL_ORDER_OUTPUT_TOPIC;
              } else {
                  outTopic = OUTPUT_TOPIC;
              }
              return outTopic;
        };

The TopicNameExtractor interface has one method, extract, which makes it suitable for using a lambda, as shown here. But remember using a concrete class has the advantage of being directly testable.

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-dynamic-output-topic-env \
  --kafka-cluster-name kafka-streams-dynamic-output-topic-cluster \
  --create-kafka-key \
  --kafka-java-properties-file ./dynamic-output-topic/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 dynamic-topic-input
confluent kafka topic create dynamic-topic-output
confluent kafka topic create special-order-output

Start a console producer:

confluent kafka topic produce dynamic-topic-input

Enter a few JSON-formatted orders:

{"id":6, "sku":"COF0003456", "name":"coffee", "quantity":1}
{"id":7, "sku":"QUA000022334", "name":"hand sanitizer", "quantity":2}

Enter Ctrl+C to exit the console producer.

Compile and run the application

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

./gradlew dynamic-output-topic:kstreams:shadowJar

Navigate into the application's home directory:

cd dynamic-output-topic/kstreams

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

java -cp ./build/libs/dynamic-output-topic-standalone.jar \
    io.confluent.developer.KafkaStreamsDynamicOutputTopic \
    ./src/main/resources/cloud.properties

Validate that you see the first order in the dynamic-topic-output topic and the second in the special-order-output topic.

confluent kafka topic consume dynamic-topic-output -b

confluent kafka topic consume special-order-output -b

Clean up

When you are finished, delete the kafka-streams-dynamic-output-topic-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 dynamic-topic-input
kafka-topics --bootstrap-server localhost:9092 --create --topic dynamic-topic-output
kafka-topics --bootstrap-server localhost:9092 --create --topic special-order-output

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

{"id":6, "sku":"COF0003456", "name":"coffee", "quantity":1}
{"id":7, "sku":"QUA000022334", "name":"hand sanitizer", "quantity":2}

./gradlew dynamic-output-topic:kstreams:shadowJar

cd dynamic-output-topic/kstreams

java -cp ./build/libs/dynamic-output-topic-standalone.jar \
    io.confluent.developer.KafkaStreamsDynamicOutputTopic \
    ./src/main/resources/local.properties

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

kafka-console-consumer --bootstrap-server localhost:9092 --topic special-order-output --from-beginning

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