title: KafkaStreams multiple streams in same application
body: <p>I'm trying to make a practical design decision based on convention and plausibility with KafkaStreams.</p>
  <p>Let's say that I have two different events that I want to place into <code>KTable</code>s. I have a producer sending these messages to a <code>KStream</code> that is listening to that topic.</p>
  <p>From what I can tell I cannot use conditional forwarding for messages using <code>KafkaStreams</code>, so if the stream is subscribe to many topics (one for each of the above messages, for example) I can only call <code>stream.to</code> on a single sink topic - otherwise, I would have to do something like call <code>foreach</code> on the stream and send messages with a <code>KProducer</code> to the sink topics.</p>
  <p>The above suggests using a single stream. I thought I could set up multiple streams in the same app, each listening to a topic, mapping and forwarding to a table sink, but everytime I try to create two instances of <code>KafkaStreams</code>, only the first initialized subscribes to its topic - the other gets a warning from the client that its topic has no subscriptions.</p>
  <p>Can I set up multiple streams in the same app? If so, are there any special requirements?</p>
  <pre class="lang-kotlin prettyprint-override"><code>    class Stream(topic: String) {
        val props: Option[Map[String, String]] = Some(TopicProps.get(topic))
        val streamsBuilder = new StreamsBuilder
        val topics = new util.ArrayList[String]
        topics.add(props.get(&quot;topic&quot;))
  
        val stream: KStream[String, String] = configureStream(streamsBuilder, topics, props.get(&quot;sink&quot;))
  
        def configureStream(builder: StreamsBuilder, topics: java.util.List[String], sink: String): KStream[String, String] = {
          builder.stream[String, String](
            topics,
            Consumed.`with`(String(), String())
          )
        }
  
        def init(): KafkaStreams = {
          val streams = new KafkaStreams(streamsBuilder.build(), KafkaConfig.streamConfig)
  
          streams.start()
  
          streams
        }
      }
  
      class Streams() {
  
        val eventStream = new Stream(&quot;first_event&quot;) //looking good!
        val eventStream2 = new Stream(&quot;second_event&quot;) // no subscribers
        //if I switch the other of these, eventStream2 is subscribed to and eventStream is dead in the water
        val streams: KafkaStreams = eventStream.init()
        val streams2: KafkaStreams = eventStream2.init()
  
      }
  </code></pre>
  <p>stream config</p>
  <pre class="lang-kotlin prettyprint-override"><code>    val streamConfig: Properties = {
          val properties = new Properties()
          properties.put(StreamsConfig.APPLICATION_ID_CONFIG, &quot;stream-application&quot;)
          properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BrokerHost)
          properties
      }
  </code></pre>
  <p>I'd also love any alternatives suggested</p>