title: Spring cloud stream not send message to Kafka when declare producer and consumer in the same application
body: <p>Currently i'm have producer and consumer configure in the same Spring Boot Application, but it very weird that the Spring cloud stream firing the message not go through the Kafka (I'm monitor the message with kafka-console-consumer), but the consumer still receive the message (using same thread as producer).</p>
  <p>And if i remove the consumerHandler (@StreamListener) in the Application, the producer successfully send the message to the Kafka.</p>
  <p>Is there any configuration for this? I need the Spring cloud stream send message to the Kafka by default.</p>
  <p>Producer and Consumer Configure:</p>
  <pre><code>@Component
  public interface NotificationProcessor {
     
      String EMAIL_NOTIFICATION = &quot;email-notification&quot;;
      @Input(EMAIL_NOTIFICATION)
      SubscribableChannel receiveEmail();
      @Output(EMAIL_NOTIFICATION)
      MessageChannel sendEmail();
  }
  </code></pre>
  <p>Here is some of my configuration:</p>
  <pre><code>spring:
    cloud:
      stream:
        kafka:
          binder:
            autoAddPartitions: true
            brokers: ${KAFKA_BROKERS:localhost:9092}
            auto-create-topics: true
            configuration:
              auto.offset.reset: latest
        bindings:
          email-notification:
            group: ${EMAIL_GROUP:email-group-notification}
            destination: ${EMAIL_TOPIC:email-notification}
            contentType: application/json
            producer:
              partitionCount: 9
            consumer:
              partitioned: true
              concurrency: 3
        instance-count: 1
        instance-index: 0
  </code></pre>
  <p>An API To Trigger Send Message:</p>
  <pre><code>@RestController
  @RequestMapping(&quot;/api&quot;)
  public class TestResource {
      private final Logger log = LoggerFactory.getLogger(TestResource.class);
  
      private final NotificationProcessor notificationProcessor;
      public TestResource(NotificationProcessor notificationProcessor) {
          this.notificationProcessor = notificationProcessor;
      }
   
      @ApiOperation(value = &quot;Test api&quot;)
      @GetMapping(value = &quot;/send-email2&quot;, produces = APPLICATION_JSON_VALUE)
      public ResponseEntity&lt;Boolean&gt; test2() {
          EmailMessage test =  EmailMessage.builder()
                  .to(Arrays.asList(Receiver.builder().email(&quot;test@nomail.com&quot;).build())
                  ).type(EContentType.JSON)
                  .build();
          log.info(&quot;send email message to kafka&quot;);
          notificationProcessor.sendEmail().send(MessageBuilder.withPayload(test).build());
          return ResponseEntity.ok(Boolean.TRUE);
      }
  }
  </code></pre>
  <p>And Consumer Handler:</p>
  <pre><code>@EnableBinding(NotificationProcessor.class)
  public class NotificationProducer {
  
      private final Logger log = LoggerFactory.getLogger(NotificationProducer.class);
  
      public NotificationProducer(){}
  
      @StreamListener(NotificationProcessor.EMAIL_NOTIFICATION)
      public void receiveEmail(@Payload Message&lt;EmailMessage&gt; message)  {
          log.info(&quot;Receive email message from kafka&quot;);
          EmailMessage emailMessage = message.getPayload();
      }
  }
  </code></pre>