title: How to listen to multiple topics by multiple StreamListener Spring Cloud Stream and Kafka stream
body: <p>I want to listen to two Kafka topics like in the code below and there are two source events that need to be handled and transform to another event.</p>
  
  <p>So I want to listen these two event in one <code>EnableBinding</code></p>
  
  <pre><code>    @EnableBinding(PartnerOrderCancelledEventImporter.Targets.class)
      public class PartnerOrderCancelledEventImporter {
  
         @StreamListener(Targets.INPUT_ORDER_CANCELLED)
          @SendTo(Targets.OUTPUT)
          public KStream&lt;?, TriggeringEvent&gt; processOrderCancelled(KStream&lt;?, OrderCancelledV1&gt; input) {
              LogInfo("PartnerOrderCancelled-OrderCancelledV1 stream started");
              LogInfo("  KafkaBrokers: " + KafkaBrokers);
  
              return input
                      .filter((key, value) -&gt; IsFFFaultAndNoRoutes(value))
                      .peek((key, value) -&gt; LogInfo("OrderCancelle", new LogObject(value)))
                      .map((key, value) -&gt; KeyValue.pair(key, new TriggeringEvent(value)));
          }
  
          @StreamListener(Targets.INPUT_ORDER_ITEM_STOCK_CHECKED)
          @SendTo(Targets.OUTPUT)
          public KStream&lt;?, TriggeringEvent&gt; processOrderItemStockChecked(KStream&lt;?, OrderItemStockCheckedV1&gt; input) {
              LogInfo("PartnerOrderCancelled-OrderItemStockCheckedV1 stream started");
              LogInfo("  KafkaBrokers: " + KafkaBrokers);
  
              return input
                      .filter((key, value) -&gt; IsItemNoStock(value))
                      .peek((key, value) -&gt; LogInfo("OrderItemStockChecke", new LogObject(value)))
                      .map((key, value) -&gt; KeyValue.pair(key, new TriggeringEvent(value)));
          }
  
  
          public interface Targets {
  
              String INPUT_ORDER_CANCELLED = "partnerOrderCancelledInputOrderCancelled";
              String INPUT_ORDER_ITEM_STOCK_CHECKED = "partnerOrderCancelledInputOrderItemStockChecked";
              String OUTPUT = "triggeringEventsOutputPartnerOrderCancelled";
  
              @Input(INPUT_ORDER_CANCELLED)
              KStream&lt;?, ?&gt; inputOrderCancelled();
  
              @Input(INPUT_ORDER_ITEM_STOCK_CHECKED)
              KStream&lt;?, ?&gt; inputOrderItemStockChecked();
  
              @Output(OUTPUT)
              KStream&lt;?, ?&gt; output();
          }
     spring.cloud.stream.bindings.triggeringEventsOutputPartnerOrderCancelled.destination=dev.comms.triggeringevents.TriggeringEvent-events-1.0
      spring.cloud.stream.bindings.partnerOrderCancelledInputOrderItemStockChecked.destination=dev.ecom.order-management-service.order-item-stock-checked-events-v1
      spring.cloud.stream.bindings.partnerOrderCancelledInputOrderItemStockChecked.group=TriggeringEvents-PartnerOrderCancelled
      spring.cloud.stream.bindings.partnerOrderCancelledInputOrderCancelled.destination=dev.ecom.order-management-service.order-cancelled-events-v1
      spring.cloud.stream.bindings.partnerOrderCancelledInputOrderCancelled.group=TriggeringEvents-PartnerOrderCancelled
  </code></pre>
  
  <p>thrown below exceptions</p>
  
  <pre><code>org.springframework.beans.factory.BeanInitializationException: Cannot setup StreamListener for public org.apache.kafka.streams.kstream.KStream com.farfetch.communication.triggeringeventsimporter.PartnerOrderCancelledEventImporter.processOrderItemStockChecked(org.apache.kafka.streams.kstream.KStream); nested exception is java.lang.reflect.UndeclaredThrowableException
  
      at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.orchestrateStreamListenerSetupMethod(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:195)
      at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.doPostProcess(StreamListenerAnnotationBeanPostProcessor.java:195)
      at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.lambda$postProcessAfterInitialization$0(StreamListenerAnnotationBeanPostProcessor.java:167)
      at java.lang.Iterable.forEach(Iterable.java:75)
      at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.injectAndPostProcessDependencies(StreamListenerAnnotationBeanPostProcessor.java:285)
      at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.afterSingletonsInstantiated(StreamListenerAnnotationBeanPostProcessor.java:105)
      。。。。。。。。。。。。。。。。。。。。。。。。。。。。。
      at org.springframework.cloud.stream.binder.kafka.streams.KStreamStreamListenerResultAdapter.adapt(KStreamStreamListenerResultAdapter.java:41)
      at org.springframework.cloud.stream.binder.kafka.streams.KStreamStreamListenerResultAdapter.adapt(KStreamStreamListenerResultAdapter.java:31)
      at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.orchestrateStreamListenerSetupMethod(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:187)
      ... 39 more
  Caused by: java.lang.reflect.InvocationTargetException
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:498)
      at org.springframework.cloud.stream.binder.kafka.streams.KStreamBoundElementFactory$KStreamWrapperHandler.invoke(KStreamBoundElementFactory.java:99)
      at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
      at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212)
      ... 43 more
  Caused by: java.lang.IllegalArgumentException: delegate already set to org.apache.kafka.streams.kstream.internals.KStreamImpl@491cafec
  </code></pre>