title: Spring boot Kafka request-reply scenario
body: <p>I am implementing POC of request/reply scenario in order to move event-based microservice stack with using Kafka.</p>
  <p>There is 2 options in spring.
  I wonder which one is better to use. <code>ReplyingKafkaTemplate</code> or <code>cloud-stream</code></p>
  <p>First is <code>ReplyingKafkaTemplate </code> which can be easily configured to have dedicated channel to reply topics for each instance.
  <code>record.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, provider.getReplyChannelName().getBytes())); </code>
  Consumer should not need to know replying topic name, just listens a topic and returns with given reply topic.</p>
  <pre class="lang-java prettyprint-override"><code>@KafkaListener(topics = &quot;${kafka.topic.concat-request}&quot;)
  @SendTo
  public ConcatReply listen(ConcatModel request) {
         .....
  }
  </code></pre>
  <p>Second option is using combination of <code>StreamListener</code>, <code>spring-integration</code> and <code>IntegrationFlows</code>. Gateway should be configured and reply topics should be filtered.</p>
  <pre class="lang-java prettyprint-override"><code>@MessagingGateway
  public interface StreamGateway {
      @Gateway(requestChannel = START, replyChannel = FILTER, replyTimeout = 5000, requestTimeout = 2000)
      String process(String payload);
  }
  @Bean
  public IntegrationFlow headerEnricherFlow() { 
      return IntegrationFlows.from(START)
              .enrichHeaders(HeaderEnricherSpec::headerChannelsToString)
              .enrichHeaders(headerEnricherSpec -&gt; headerEnricherSpec.header(Channels.INSTANCE_ID ,instanceUUID)) 
              .channel(Channels.REQUEST)
              .get();
  }
  @Bean
  public IntegrationFlow replyFiltererFlow() {
      return IntegrationFlows.from(GatewayChannels.REPLY)
              .filter(Message.class, message -&gt; Channels.INSTANCE_ID.equals(message.getHeaders().get(&quot;instanceId&quot;)) )
              .channel(FILTER)
              .get();
  }
  </code></pre>
  <p>Building reply</p>
  <pre class="lang-java prettyprint-override"><code>@StreamListener(Channels.REQUEST)
  @SendTo(Channels.REPLY)
  public Message&lt;?&gt; process(Message&lt;String&gt; request) { 
  </code></pre>
  <p>Specifying reply channel is mandatory. So receieved reply topics are filtered according to instanceID which a kind of workaround (might bloat up the network). On the other hand, DLQ scenario is enabled with adding</p>
  <pre class="lang-yaml prettyprint-override"><code>              consumer:
                  enableDlq: true
  </code></pre>
  <p>Using spring cloud streams looks promising in terms of interoperability with RabbitMQ and other features, but not officially supports request reply scenario right away. Issue is still open, not rejected also. (<a href="https://github.com/spring-cloud/spring-cloud-stream/issues/1800" rel="nofollow noreferrer">https://github.com/spring-cloud/spring-cloud-stream/issues/1800</a>)</p>
  <p>Any suggestions are welcomed.</p>