title: Forwarding spring @KafkaListener result with @SendTo
body: <p>I am creating a message processor that receives an event from one kafka topic, processes it and forwards the result to another topic.</p>
  
  <p>I created a <code>@KafkaListener</code> method with <code>@SendTo</code> which works fine until I want to take control of the key generation of the outgoing message.</p>
  
  <p>The <a href="https://docs.spring.io/spring-kafka/docs/2.2.4.RELEASE/reference/#annotation-send-to" rel="nofollow noreferrer">documentation (2.2.4.RELEASE)</a> suggest to create a bean by sub-classing <code>KafkaTemplate</code> and override it's <code>send(String topic, String data)</code> method.</p>
  
  <p>Unfortunately this does not work because that method is not called in my case. <code>send(Message&lt;?&gt; message)</code> is called on the other hand but that does not help. After a short debugging it turned out that <code>MessagingMessageListenerAdapter</code> calls this method in case the input is instance of <code>org.springframework.messaging.Message</code> and the result is not a <code>List</code>. Unfortunately the <code>RecordMessagingMessageListenerAdapter</code> always transforms the <em>input</em> to a <code>Message</code>.
  Am I using this annotation combination for something that was not the intention of the authors of spring kafka, is this a bug or is the documentation wrong?</p>
  
  <p>Besides it is quite annoying that spring boot auto configuration works only if I don't create my own <code>KafkaTemplate</code> bean.
  If I create that overridden template then I have to create the KafkaListenerContainerFactory myself and set the replying template to make <code>@SendTo</code> work again.</p>
  
  <p>Here is my example code. As simple as it can be.</p>
  
  <pre class="lang-java prettyprint-override"><code>@SpringBootApplication
  @Slf4j
  public class SpringKafkaExampleApplication {
  
      public static void main(String[] args) {
          SpringApplication.run(SpringKafkaExampleApplication.class, args);
      }
  
      @KafkaListener(topics = "${example.topics.input}")
      @SendTo("${example.topics.output}")
      public String process(final byte[] payload) {
          String message = new String(payload, StandardCharsets.UTF_8);
          log.info(message);
          return message;
      }
  
  /*
      //To set my custom KafkaTemplate as replyTemplate
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, byte[]&gt; kafkaListenerContainerFactory(KafkaTemplate&lt;String, String&gt; kafkaTemplate,
                                                                                                   ConsumerFactory&lt;String, byte[]&gt; consumerFactory) {
          ConcurrentKafkaListenerContainerFactory&lt;String, byte[]&gt; factory =
                  new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory);
          factory.setReplyTemplate(kafkaTemplate);
          return factory;
      }
  
      //My KafkaTemplate with overridden send(topic, data) method
      @Bean
      public KafkaTemplate&lt;String, String&gt; kafkaTempate(ProducerFactory&lt;String, String&gt; producerFactory) {
          return new KafkaTemplate&lt;String, String&gt;(producerFactory) {
  
              @Override
              public ListenableFuture&lt;SendResult&lt;String, String&gt;&gt; send(String topic, String data) {
                  return super.send(topic, "some_generated_key", data);
              }
          };
      }
      */
  }
  
  </code></pre>
  
  <p><strong>UPDATE</strong></p>
  
  <h3>stack trace ending in send(Message)</h3>
  
  <pre><code>send:215, KafkaTemplate (org.springframework.kafka.core)
  sendReplyForMessageSource:449, MessagingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
  sendSingleResult:416, MessagingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
  sendResponse:402, MessagingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
  handleResult:324, MessagingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
  onMessage:81, RecordMessagingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
  onMessage:50, RecordMessagingMessageListenerAdapter (org.springframework.kafka.listener.adapter)
  </code></pre>
  
  <h3>RecordMessagingMessageListenerAdapter</h3>
  
  <p>Here the received record is transformed to a Message object.</p>
  
  <pre class="lang-java prettyprint-override"><code>    @Override
      public void onMessage(ConsumerRecord&lt;K, V&gt; record, Acknowledgment acknowledgment, Consumer&lt;?, ?&gt; consumer) {
          Message&lt;?&gt; message = toMessagingMessage(record, acknowledgment, consumer);
          if (logger.isDebugEnabled()) {
              logger.debug("Processing [" + message + "]");
          }
          try {
              Object result = invokeHandler(record, acknowledgment, message, consumer);
              if (result != null) {
                  handleResult(result, record, message);
              }
          }
  </code></pre>
  
  <h3>MessagingMessageListenerAdapter</h3>
  
  <p><code>String</code> is returned by the KafkaListener method so <code>sendSingleResult(result, topic, source)</code> will be called.</p>
  
  <pre class="lang-java prettyprint-override"><code>    protected void sendResponse(Object result, String topic, @Nullable Object source, boolean messageReturnType) {
          if (!messageReturnType &amp;&amp; topic == null) {
              if (this.logger.isDebugEnabled()) {
                  this.logger.debug("No replyTopic to handle the reply: " + result);
              }
          }
          else if (result instanceof Message) {
              this.replyTemplate.send((Message&lt;?&gt;) result);
          }
          else {
              if (result instanceof Collection) {
                  ((Collection&lt;V&gt;) result).forEach(v -&gt; {
                      if (v instanceof Message) {
                          this.replyTemplate.send((Message&lt;?&gt;) v);
                      }
                      else {
                          this.replyTemplate.send(topic, v);
                      }
                  });
              }
              else {
                  sendSingleResult(result, topic, source);
              }
          }
      }
  </code></pre>
  
  <pre class="lang-java prettyprint-override"><code>    private void sendSingleResult(Object result, String topic, @Nullable Object source) {
          byte[] correlationId = null;
          boolean sourceIsMessage = source instanceof Message;
          if (sourceIsMessage
                  &amp;&amp; ((Message&lt;?&gt;) source).getHeaders().get(KafkaHeaders.CORRELATION_ID) != null) {
              correlationId = ((Message&lt;?&gt;) source).getHeaders().get(KafkaHeaders.CORRELATION_ID, byte[].class);
          }
          if (sourceIsMessage) {
              sendReplyForMessageSource(result, topic, source, correlationId);
          }
          else {
              this.replyTemplate.send(topic, result);
          }
      }
  
      @SuppressWarnings("unchecked")
      private void sendReplyForMessageSource(Object result, String topic, Object source, byte[] correlationId) {
          MessageBuilder&lt;Object&gt; builder = MessageBuilder.withPayload(result)
                  .setHeader(KafkaHeaders.TOPIC, topic);
          if (this.replyHeadersConfigurer != null) {
              Map&lt;String, Object&gt; headersToCopy = ((Message&lt;?&gt;) source).getHeaders().entrySet().stream()
                  .filter(e -&gt; {
                      String key = e.getKey();
                      return !key.equals(MessageHeaders.ID) &amp;&amp; !key.equals(MessageHeaders.TIMESTAMP)
                              &amp;&amp; !key.equals(KafkaHeaders.CORRELATION_ID)
                              &amp;&amp; !key.startsWith(KafkaHeaders.RECEIVED);
                  })
                  .filter(e -&gt; this.replyHeadersConfigurer.shouldCopy(e.getKey(), e.getValue()))
                  .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
              if (headersToCopy.size() &gt; 0) {
                  builder.copyHeaders(headersToCopy);
              }
              headersToCopy = this.replyHeadersConfigurer.additionalHeaders();
              if (!ObjectUtils.isEmpty(headersToCopy)) {
                  builder.copyHeaders(headersToCopy);
              }
          }
          if (correlationId != null) {
              builder.setHeader(KafkaHeaders.CORRELATION_ID, correlationId);
          }
          setPartition(builder, ((Message&lt;?&gt;) source));
          this.replyTemplate.send(builder.build());
      }
  </code></pre>
  
  <p><code>source</code> is a Message now -> <code>sendReplyForMessageSource</code> will be called.</p>