title: How to use multiple transformers using the same topic for kafka streams?
body: <p>I need to parse complex messages on kafka using multiple transformers. Each transformer parses a part of the message and edits the message by filling some attributes on the message. In the end the fully parsed message is stored in the database using a Kafka consumer. Currently, I'm doing this:</p>
  <pre><code>streamsBuilder.stream(Topic.A, someConsumer)
         \\ filters messages that have unparsed parts of type X
         .filter(filterX)
         \\ transformer that edits the message and produces new Topic.E messages
         .transform(ParseXandProduceE::new)
         .to(Topic.A, someProducer)
  
  streamsBuilder.stream(Topic.A, someConsumer)
         \\ filters messages that have unparsed parts of type Y
         .filter(filterY)
         \\ transformer that edits the message and produces new Topic.F messages
         .transform(ParseYandProduceF::new)
         .to(Topic.A, someProducer)
  </code></pre>
  <p>a Transformer looks like:</p>
  <pre><code>class ParseXandProduceE implements Transformer&lt;...&gt; {
      @Override
      public KeyValue&lt;String, Message&gt; transform (String key, Message message) {
             message.x = parse(message.rawX);
             context.forward(newKey, message.x, Topic.E);
             return KeyValue.pair(key, message);
      }
  }
  </code></pre>
  <p>However, this is cumbersome, the same messages flow multiple times through these streams.
  Additionally, there is a consumer that stores messages of <code>topic.A</code> in the database. Messages are currently stored multiple times, before each transformation and after each transformation. It is necessary to store each message once.</p>
  <p>The following could work, but seems unfavorable since each block of filter+transform could have been put cleanly in its own separate class:</p>
  <pre><code>streamsBuilder.stream(Topic.A, someConsumer)
         \\ transformer that filters and edits the message and produces new Topic.E + Topic.F messages
         .transform(someTransformer)
         .to(Topic.B, someProducer)
  </code></pre>
  <p>and make the persistence consumer listen to <code>Topic.B</code>.</p>
  <p>Is the latter proposed solution the way to go, or is there some other way to achieve the same result? Maybe with a complete Topology configuration of Sources and Sinks? If so, what would that look like for this scenario?</p>