Forwarding spring @KafkaListener result with @SendTo

Question

I am creating a message processor that receives an event from one kafka topic, processes it and forwards the result to another topic.

I created a @KafkaListener method with @SendTo which works fine until I want to take control of the key generation of the outgoing message.

The documentation (2.2.4.RELEASE) suggest to create a bean by sub-classing KafkaTemplate and override it's send(String topic, String data) method.

Unfortunately this does not work because that method is not called in my case. send(Message<?> message) is called on the other hand but that does not help. After a short debugging it turned out that MessagingMessageListenerAdapter calls this method in case the input is instance of org.springframework.messaging.Message and the result is not a List. Unfortunately the RecordMessagingMessageListenerAdapter always transforms the input to a Message. 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?

Besides it is quite annoying that spring boot auto configuration works only if I don't create my own KafkaTemplate bean. If I create that overridden template then I have to create the KafkaListenerContainerFactory myself and set the replying template to make @SendTo work again.

Here is my example code. As simple as it can be.

@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<String, byte[]> kafkaListenerContainerFactory(KafkaTemplate<String, String> kafkaTemplate,
                                                                                                 ConsumerFactory<String, byte[]> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, byte[]> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setReplyTemplate(kafkaTemplate);
        return factory;
    }

    //My KafkaTemplate with overridden send(topic, data) method
    @Bean
    public KafkaTemplate<String, String> kafkaTempate(ProducerFactory<String, String> producerFactory) {
        return new KafkaTemplate<String, String>(producerFactory) {

            @Override
            public ListenableFuture<SendResult<String, String>> send(String topic, String data) {
                return super.send(topic, "some_generated_key", data);
            }
        };
    }
    */
}

UPDATE

stack trace ending in send(Message)

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)

RecordMessagingMessageListenerAdapter

Here the received record is transformed to a Message object.

    @Override
    public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
        Message<?> 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);
            }
        }

MessagingMessageListenerAdapter

String is returned by the KafkaListener method so sendSingleResult(result, topic, source) will be called.

    protected void sendResponse(Object result, String topic, @Nullable Object source, boolean messageReturnType) {
        if (!messageReturnType && 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<?>) result);
        }
        else {
            if (result instanceof Collection) {
                ((Collection<V>) result).forEach(v -> {
                    if (v instanceof Message) {
                        this.replyTemplate.send((Message<?>) v);
                    }
                    else {
                        this.replyTemplate.send(topic, v);
                    }
                });
            }
            else {
                sendSingleResult(result, topic, source);
            }
        }
    }

    private void sendSingleResult(Object result, String topic, @Nullable Object source) {
        byte[] correlationId = null;
        boolean sourceIsMessage = source instanceof Message;
        if (sourceIsMessage
                && ((Message<?>) source).getHeaders().get(KafkaHeaders.CORRELATION_ID) != null) {
            correlationId = ((Message<?>) 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<Object> builder = MessageBuilder.withPayload(result)
                .setHeader(KafkaHeaders.TOPIC, topic);
        if (this.replyHeadersConfigurer != null) {
            Map<String, Object> headersToCopy = ((Message<?>) source).getHeaders().entrySet().stream()
                .filter(e -> {
                    String key = e.getKey();
                    return !key.equals(MessageHeaders.ID) && !key.equals(MessageHeaders.TIMESTAMP)
                            && !key.equals(KafkaHeaders.CORRELATION_ID)
                            && !key.startsWith(KafkaHeaders.RECEIVED);
                })
                .filter(e -> this.replyHeadersConfigurer.shouldCopy(e.getKey(), e.getValue()))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
            if (headersToCopy.size() > 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<?>) source));
        this.replyTemplate.send(builder.build());
    }

source is a Message now -> sendReplyForMessageSource will be called.

Answer 1

After a short debugging it turned out that MessagingMessageListenerAdapter calls this method in case the input is instance of org.springframework.messaging.Message

That is not correct; it's called when the listener method returns a Message<?> (or Collection<Message<?>>).

Code:

protected void sendResponse(Object result, String topic, @Nullable Object source, boolean messageReturnType) {
    if (!messageReturnType && 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<?>) result);
    }
    else {
        if (result instanceof Collection) {
            ((Collection<V>) result).forEach(v -> {
                if (v instanceof Message) {
                    this.replyTemplate.send((Message<?>) v);
                }
                else {
                    this.replyTemplate.send(topic, v);
                }
            });
        }
        else {
            sendSingleResult(result, topic, source);
        }
    }
}

The simplest way to customize the outbound key is to change your method to return Message<String>. Scroll down from that documentation link to ...

If the listener method returns Message or Collection>, the listener method is responsible for setting up the message headers for the reply. For example, when handling a request from a ReplyingKafkaTemplate, you might do the following:

@KafkaListener(id = "messageReturned", topics = "someTopic")
public Message<?> listen(String in, @Header(KafkaHeaders.REPLY_TOPIC) byte[] replyTo,
        @Header(KafkaHeaders.CORRELATION_ID) byte[] correlation) {

    return MessageBuilder.withPayload(in.toUpperCase())
            .setHeader(KafkaHeaders.TOPIC, replyTo)
            .setHeader(KafkaHeaders.MESSAGE_KEY, 42)
            .setHeader(KafkaHeaders.CORRELATION_ID, correlation)
            .setHeader("someOtherHeader", "someValue")
            .build();
}

Besides it is quite annoying that spring boot auto configuration works only if I don't create my own KafkaTemplate bean.

In order for boot to wire in the reply template, it must be declared as KafkaTemplate<Object, Object>.