title: How to process dynamically in kafka streams and send to different topic
body: <p>I am creating a stream processing app. It should create a kafka stream connection. When a message arrives following are the required things I want to do:</p>
  
  <ul>
  <li>check the type of message</li>
  <li>process the message by calling a specificTypeProcessing Service</li>
  <li><p>end it to specific topics that are decided on the basis of the message type</p>
  
  <pre><code>public java.util.function.Consumer&lt;KStream&lt;String, String&gt;&gt; process() {
      String topic;
      return input -&gt;
              input.map((key, value) -&gt; {
                  //check type and ask object from factory
                  try {
                      JSONObject msg = Util.getObjectMapper().readValue(value, JSONObject.class);
                      String type = msg.get("type").toString();
                      if(type.equalsIgnoreCase("test")){
                          //processing started
                          msgTypeHandlerFactory
                                  .createInstance(type)
                                  .enrichAndRelay(msg);
                          System.out.println("IN");
                      }
                      else{
                          input.to("notStream");
                          System.out.println("OUT");
                      }
  
                  } catch (JsonProcessingException e) {
                      e.printStackTrace();
                  } catch (Exception e) {
                      e.printStackTrace();
                  }
                  return KeyValue.pair(key, value);
  
              })
              .to("output_topic");
  
  }
  </code></pre>
  
  <p>The issue with the above code is I'm using map function which grants me the ability to use the .to() function to send the stream.
  What I want is to check every message for type and then process, send to another stream accordingly.
  For this, I should use foreach function which doesn't give me .to() function so I have to create another Kafka Producer to do the job.</p></li>
  </ul>
  
  <p>Requirement:</p>
  
  <ol>
  <li>Each msg should be processed and sent before processing the next msg with the help of stream function and not using another kafka producer</li>
  <li>If the requirement one is achieved then I should be able to send msg to topics that will be decided dynamically according to type.</li>
  </ol>