Kafka Streams - Send on different topics depending on Streams Data

You can use branch method in order to split your stream. This method takes predicates for splitting the source stream into several streams.

The code below is taken from kafka-streams-examples:

KStream<String, OrderValue>[] forks = ordersWithTotals.branch(
    (id, orderValue) -> orderValue.getValue() >= FRAUD_LIMIT,
    (id, orderValue) -> orderValue.getValue() < FRAUD_LIMIT);

forks[0].mapValues(
    orderValue -> new OrderValidation(orderValue.getOrder().getId(), FRAUD_CHECK, FAIL))
    .to(ORDER_VALIDATIONS.name(), Produced
        .with(ORDER_VALIDATIONS.keySerde(), ORDER_VALIDATIONS.valueSerde()));

forks[1].mapValues(
    orderValue -> new OrderValidation(orderValue.getOrder().getId(), FRAUD_CHECK, PASS))
    .to(ORDER_VALIDATIONS.name(), Produced
  .with(ORDER_VALIDATIONS.keySerde(), ORDER_VALIDATIONS.valueSerde()));

Another possibility is routing the event dynamically using a TopicNameExtractor:

https://www.confluent.io/blog/putting-events-in-their-place-with-dynamic-routing

you would need to have created the topics in advance though,

val outputTopic: TopicNameExtractor[String, String] = (_, value: String, _) => defineOutputTopic(value)

builder
  .stream[String, String](inputTopic)
  .to(outputTopic)

and defineOutputTopic can return one of a defined set of topics given the value (or key or record context for that matter). PD: sorry for the scala code, in the link there is a Java example.

The original KStream.branch method is inconvenient because of mixed arrays and generics, and because it forces one to use 'magic numbers' to extract the right branch from the result (see e.g. KAFKA-5488 issue). Starting from spring-kafka 2.2.4, KafkaStreamBrancher class is available. With it, more convenient branching is possible:

        
new KafkaStreamBrancher<String, String>()
    .branch((key, value) -> value.contains("A"), ks->ks.to("A"))
    .branch((key, value) -> value.contains("B"), ks->ks.to("B"))
    .defaultBranch(ks->ks.to("C"))
    .onTopOf(builder.stream("source"))
    //onTopOf returns the provided stream so we can continue with method chaining 
    //and do something more with the original stream

There is also KIP-418, so a there is also a chance that branching will be improved in Kafka itself in further releases.