title: Multiple StreamListeners with Spring Cloud Stream connected to Kafka
body: <p>In a Spring Boot app using Spring Cloud Stream connecting to Kafka, I'm trying to set up two separate stream listener methods:</p>
  
  <ul>
  <li>One reads from topics "t1" and "t2" as KTables, re-partitioning using a different key in one, then joining to data from the other</li>
  <li>The other reads from an unrelated topic, "t3", as a KStream.</li>
  </ul>
  
  <p>Because the first listener does some joining and aggregating, some topics are created automatically, e.g. "test-1-KTABLE-AGGREGATE-STATE-STORE-0000000007-repartition-0".  (Not sure if this is related to the problem or not.)</p>
  
  <p>When I set up the code by having two separate methods annotated with <code>@StreamListener</code>, I get the error below when the Spring Boot app starts:</p>
  
  <pre><code>Exception in thread "test-d44cb424-7575-4f5f-b148-afad034c93f4-StreamThread-2" java.lang.IllegalArgumentException: Assigned partition t1-0 for non-subscribed topic regex pattern; subscription pattern is t3
      at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignFromSubscribed(SubscriptionState.java:195)
      at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:225)
      at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:367)
      at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316)
      at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295)
      at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1146)
      at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1111)
      at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:848)
      at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:805)
      at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:771)
      at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:741)
  </code></pre>
  
  <p>I think the important part is: "Assigned partition <strong>t1-0</strong> for non-subscribed topic regex pattern; subscription pattern is <strong>t3</strong>". These are the two unrelated topics, so as far as I can see nothing related to t3 should be subscribing to anything related to t1.  The exact topic which causes the problem also changes intermittently: sometimes it's one of the automatically generated topics which is mentioned, rather than t1 itself.</p>
  
  <p>Here is how the two stream listeners are set up (in Kotlin):</p>
  
  <pre><code>@StreamListener
  fun listenerForT1AndT2(
          @Input("t1") t1KTable: KTable&lt;String, T1Obj&gt;,
          @Input("t2") t2KTable: KTable&lt;String, T2Obj&gt;) {
  
      t2KTable
          .groupBy(...)
          .aggregate(
                  { ... },
                  { ... },
                  { ... },
                  Materialized.with(Serdes.String(), JsonSerde(SomeObj::class.java)))
          .join(t1KTable,
                  { ... },
                  Materialized.`as`&lt;String, SomeObj, KeyValueStore&lt;Bytes, ByteArray&gt;&gt;("test")
                          .withKeySerde(Serdes.String())
                          .withValueSerde(JsonSerde(SomeObj::class.java)))
  }
  
  @StreamListener
  fun listenerForT3(@Input("t3") t3KStream: KStream&lt;String, T3Obj&gt;) {
      events.map { ... }
  }
  </code></pre>
  
  <p>However, when I set up my code by having just one method annotated with <code>@StreamListener</code>, and take parameters for all three topics, everything works fine, e.g.</p>
  
  <pre><code>@StreamListener
  fun compositeListener(
          @Input("t1") t1KTable: KTable&lt;String, T1Obj&gt;,
          @Input("t2") t2KTable: KTable&lt;String, T2Obj&gt;,
          @Input("t3") t3KStream: KStream&lt;String, T3Obj&gt;) {
      ...
  }
  </code></pre>
  
  <p>But I don't think it's right that I can only have one <code>@StreamListener</code> method.</p>
  
  <p>I know that there is <a href="https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#_using_streamlistener_for_content_based_routing" rel="nofollow noreferrer">content-based routing</a> for adding conditions to the <code>StreamListener</code> annotation, but if the methods define the input channels then I'm not sure if I need to be using this here - I'd have thought the use of the <code>@Input</code> annotations on the method parameters would be enough to tell the system which channels (and therefore which Kafka topics) to bind to? If I <strong>do</strong> need to use content-based routing, how can I apply it here to have each method receive only the items from the relevant topic(s)?</p>
  
  <p>I've also tried separating out the two listener methods into two separate classes, each of which has <code>@EnableBinding</code> for only the interface it's interested in (i.e. one interface for t1 and t2, and a separate interface for t3), but that doesn't help.</p>
  
  <p>Everything else I've found related to this error message, e.g. <a href="https://stackoverflow.com/questions/47997066/kafka-streams-use-the-same-application-id-to-consume-from-multiple-topics">here</a>, is about having multiple app instances, but in my case there's only one Spring Boot app instance.</p>