Search code examples
javaapache-kafkaapache-kafka-streams

send message to topic based on message key using kafka streams

I want to be able to send all records in a Kafkastream to a different topic based on the key of the message key. Ex. A stream in Kafka contains name as key and record as value. I want to fan out these records to different topic based on the key of the record

data : (jhon -> {jhonsRecord}),(sean -> {seansRecord}),(mary -> {marysRecord}),(jhon -> {jhonsRecord2}), expected

  • topic1 :name: jhon ->(jhon -> {jhonsRecord}),(jhon -> {jhonsRecord2})
  • topic2 :sean-> (sean -> {seansRecord})
  • topic3 :mary -> (mary -> {marysRecord})

Below is the way I am doing this right now, but since the list of names is hudge this is slow. Plus even if there are records a few names, I need to traverse the entire list Please suggest a fix

    for( String name : names )
    {
        recordsByName.filterNot(( k, v ) -> k.equalsIgnoreCase(name)).to(name);
    }
Solution

  • I think you should use KStream::to(final TopicNameExtractor<K, V> topicExtractor) function. It gives you ability to calculate name of the topic for each message.

    Sample code:

    final KStream<String, String> stream = ???;
stream.to((key, value, recordContext) -> key);