title: Using kafka streams to segregate messages
body: <p>I have a setup where each kafka message will contain a "sender" field. All these message are sent to a single topic. </p>
  
  <p>Is there a way to segregate these messages at the consumer side? I would like sender specific consumer that will read all messages pertaining to that sender alone. </p>
  
  <p>Should I be using Kafka Streams to achieve this? I am new to Kafka Streams, any advice guidance will be helpful.</p>
  
  <pre><code>public class KafkaStreams3 {
  
  public static void main(String[] args) throws JSONException {       
  
      Properties props = new Properties();
      props.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafkastreams1");
      props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  
      props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  
      final Serde &lt; String &gt; stringSerde = Serdes.String();
  
      Properties kafkaProperties = new Properties();
      kafkaProperties.put("key.serializer",
              "org.apache.kafka.common.serialization.StringSerializer");
      kafkaProperties.put("value.serializer",
              "org.apache.kafka.common.serialization.StringSerializer");
  
      kafkaProperties.put("bootstrap.servers", "localhost:9092");
  
      KafkaProducer&lt;String, String&gt; producer = new KafkaProducer&lt;String, String&gt;(kafkaProperties);
  
  
  
      KStreamBuilder builder = new KStreamBuilder();
  
     KStream&lt;String, String&gt; source = builder.stream(stringSerde, stringSerde, "topic1");
  
  
      KStream&lt;String, String&gt; s1 = source.map(new KeyValueMapper&lt;String, String, KeyValue&lt;String, String&gt;&gt;() {
          @Override
          public KeyValue&lt;String, String&gt; apply(String dummy, String record) {
              JSONObject jsonObject;
  
              try {
                  jsonObject = new JSONObject(record);
                  return new KeyValue&lt;String,String&gt;(jsonObject.get("sender").toString(), record);
              } catch (JSONException e) {
                  e.printStackTrace();
                  return new KeyValue&lt;&gt;(record, record);
              }
  
          }
        });
  
      s1.print();
  
      s1.foreach(new ForeachAction&lt;String, String&gt;() {
  
          @Override
          public void apply(String key, String value) {
              ProducerRecord&lt;String, String&gt; data1 = new ProducerRecord&lt;String, String&gt;(
                      key, key, value);
              producer.send(data1);
  
          }
  
      });
  
      KafkaStreams streams = new KafkaStreams(builder, props);
  
      streams.start();
  
      Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
          @Override
          public void run() {
            streams.close();
            producer.close();
          }
        }));
  
  }
  </code></pre>
  
  <p>}</p>