title: Spring Cloud Stream Kafka Multiple Binding
body: <p>I am using Spring Cloud Stream Kafka binder to consume messages from Kafka. I am able to make my sample work with a single Kafka Binder as below</p>
  <pre><code>spring:
    cloud:
      stream:
        kafka:
          binder:
            consumer-properties: {enable.auto.commit: true}
            auto-create-topics: false
            brokers: &lt;broker url&gt;
        bindings:
          consumer:
            destination: some-topic
            group: testconsumergroup
            consumer:
              concurrency: 1
              valueSerde: JsonSerde
          producer:
            destination: some-other-topic
            producer:
              valueSerde: JsonSerde
  </code></pre>
  <p>Note that both the bindings are to the same Kafka Broker here. However, I have a situation where I need to publish to a topic in some Kafka Cluster and also consume from another topic in a different Kafka Cluster. How should I change my configuration to be able to bind to different Kafka Clusters?</p>
  <p>I tried something like this</p>
  <pre><code>spring:
    cloud:
      stream:
        binders:
          defaultbinder:
            type: kafka
            environment:
              spring.cloud.stream.kafka.streams.binder.brokers: &lt;cluster1-brokers&gt;
          kafka1:
            type: kafka
            environment:
              spring.cloud.stream.kafka.streams.binder.brokers: &lt;cluster2-brokers&gt;
        bindings:
          consumer:
            binder: kafka1
            destination: some-topic
            group: testconsumergroup
            consumer:
              concurrency: 1
              valueSerde: JsonSerde
          producer:
            binder: defaultbinder
            destination: some-topic
            producer:
              valueSerde: JsonSerde      
        kafka:
          binder:
            consumer-properties: {enable.auto.commit: true}
            auto-create-topics: false
            brokers: &lt;cluster1-brokers&gt;
  </code></pre>
  <p>and</p>
  <pre><code>spring:
    cloud:
      stream:
        binders:
          defaultbinder:
            type: kafka
            environment:
              spring.cloud.stream.kafka.streams.binder.brokers: &lt;cluster1-brokers&gt;
          kafka1:
            type: kafka
            environment:
              spring.cloud.stream.kafka.streams.binder.brokers: &lt;cluster2-brokers&gt;
        kafka:
          bindings:
            consumer:
              binder: kafka1
              destination: some-topic
              group: testconsumergroup
              consumer:
                concurrency: 1
                valueSerde: JsonSerde
            producer:
              binder: defaultbinder
              destination: some-topic
              producer:
                valueSerde: JsonSerde      
        kafka:
          binder:
            consumer-properties: {enable.auto.commit: true}
            auto-create-topics: false
            brokers: &lt;cluster1-brokers&gt;
  </code></pre>
  <p>But both of them dont seem to work.
  The first configuration seems to be invalid.
  For the second configuration I get the below error</p>
  <pre><code>Caused by: java.lang.IllegalStateException: A default binder has been requested, but there is more than one binder available for 'org.springframework.cloud.stream.messaging.DirectWithAttributesChannel' : kafka1,defaultbinder, and no default binder has been set.
  </code></pre>
  <p><strong>I am using the dependency 'org.springframework.cloud:spring-cloud-starter-stream-kafka:3.0.1.RELEASE' and Spring Boot 2.2.6</strong></p>
  <p>Please let me know how to configure multiple bindings for Kafka using Spring Cloud Stream</p>
  <p><strong>Update</strong></p>
  <p>Tried this configuration below</p>
  <pre><code>spring:
    cloud:
      stream:
        binders:
          kafka2:
            type: kafka
            environment:
              spring.cloud.stream.kafka.binder.brokers: &lt;cluster2-brokers&gt;
          kafka1:
            type: kafka
            environment:
              spring.cloud.stream.kafka.binder.brokers: &lt;cluster1-brokers&gt;
        bindings:
          consumer:
            destination: &lt;some-topic&gt;
            binder: kafka1
            group: testconsumergroup
            content-type: application/json
            nativeEncoding: true
            consumer:
              concurrency: 1
              valueSerde: JsonSerde
          producer:
            destination: some-topic
            binder: kafka2
            contentType: application/json
            nativeEncoding: true
            producer:
              valueSerde: JsonSerde
  </code></pre>
  <p>The Message Streams and EventHubBinding is as follows</p>
  <pre><code>public interface MessageStreams {
    String PRODUCER = &quot;producer&quot;;
    String CONSUMER = &quot;consumer;
  
    @Output(PRODUCER)
    MessageChannel producerChannel();
  
    @Input(CONSUMER)
    SubscribableChannel consumerChannel()
  }
  
  @EnableBinding(MessageStreams.class)
  public class EventHubStreamsConfiguration {
  }
  </code></pre>
  <p>My Producer class looks like below</p>
  <pre><code>@Component
  @Slf4j
  public class EventPublisher {
    private final MessageStreams messageStreams;
  
    public EventPublisher(MessageStreams messageStreams) {
      this.messageStreams = messageStreams;
    }
  
    public boolean publish(CustomMessage event) {
      MessageChannel messageChannel = getChannel();
      MessageBuilder messageBuilder = MessageBuilder.withPayload(event);
      boolean messageSent = messageChannel.send(messageBuilder.build());
      return messageSent;
    }
  
    protected MessageChannel getChannel() {
      return messageStreams.producerChannel();
    }
  }
  </code></pre>
  <p>And Consumer class looks like below</p>
  <pre><code>@Component
  @Slf4j
  public class EventHandler {
    private final MessageStreams messageStreams;
  
    public EventHandler(MessageStreams messageStreams) {
      this.messageStreams = messageStreams;
    }
  
    @StreamListener(MessageStreams.CONSUMER)
    public void handleEvent(Message&lt;CustomMessage&gt; message) throws Exception 
    {
      // process the event
    }
  
    @Override
    @ServiceActivator(inputChannel = &quot;some-topic.testconsumergroup.errors&quot;)
    protected void handleError(ErrorMessage errorMessage) throws Exception {
      // handle error;
    }
  }
  </code></pre>
  <p>I am getting the below error while trying to publish and consume the messages from my test.</p>
  <pre><code>Dispatcher has no subscribers for channel 'application.producer'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[104], headers={contentType=application/json, timestamp=1593517340422}]
  </code></pre>
  <p>Am i missing anything? For a single cluster, i am able to publish and consume messages. The issue is only happening with multiple cluster bindings</p>