Пересылка пружины @KafkaListener результат с @SendTo - PullRequest
Новая
       1

Пересылка пружины @KafkaListener результат с @SendTo

0 голосов
/ 29 марта 2019

Я создаю обработчик сообщений, который получает событие из одной темы kafka, обрабатывает его и перенаправляет результат в другую тему.

Я создал метод @KafkaListener с @SendTo, который работает нормально, пока яхотите получить контроль над генерацией ключа исходящего сообщения.

Документация (2.2.4.RELEASE) предлагает создать компонент с помощью подкласса KafkaTemplate и переопределить его send(String topic, String data) method.

К сожалению, это не работает, потому что этот метод не вызывается в моем случае.send(Message<?> message) вызывается с другой стороны, но это не помогает.После короткой отладки выяснилось, что MessagingMessageListenerAdapter вызывает этот метод в случае, если вход является экземпляром org.springframework.messaging.Message, а результат не является List.К сожалению, RecordMessagingMessageListenerAdapter всегда преобразует вход в Message.Использую ли я эту комбинацию аннотаций для чего-то, что не было задумано авторами spring kafka, это ошибка или документация неправильная?

Кроме того, довольно неприятно, что автоматическая конфигурация весенней загрузки работает, только если яне создавай мой собственный KafkaTemplate боб.Если я создаю этот переопределенный шаблон, я должен сам создать KafkaListenerContainerFactory и настроить отвечающий шаблон так, чтобы @SendTo снова работал.

Вот мой пример кода.Как можно проще. 

@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);
            }
        };
    }
    */
}

ОБНОВЛЕНИЕ

трассировка стека, заканчивающаяся отправкой (сообщение) 

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

Здесь полученная запись преобразуется в объект сообщения. 

    @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 возвращается методом KafkaListener, поэтому будет вызываться sendSingleResult(result, topic, source). 

    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 - это сообщение сейчас -> sendReplyForMessageSource будет вызываться.

1 Ответ

0 голосов
/ 29 марта 2019

После короткой отладки оказалось, что MessagingMessageListenerAdapter вызывает этот метод в случае, если вход является экземпляром org.springframework.messaging.Message

Это не правильно; вызывается, когда метод слушателя возвращает a Message<?> (или Collection<Message<?>>).

Код: 

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);
        }
    }
}

Самый простой способ настроить исходящий ключ - это изменить свой метод на возврат Message<String>. Прокрутите вниз от этой документации к ссылке ...

Если метод слушателя возвращает сообщение или коллекцию>, метод слушателя отвечает за настройку заголовков сообщений для ответа. Например, при обработке запроса от ReplyingKafkaTemplate вы можете сделать следующее: 

@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();
}

Кроме того, довольно неприятно, что автоматическая конфигурация весенней загрузки работает, только если я не создаю свой собственный компонент KafkaTemplate.

Для загрузки в шаблон ответа, он должен быть объявлен как KafkaTemplate<Object, Object>.

...