如何在使用Spring Kafka非阻塞重试向DLT发送消息时添加自定义标头？

Question

我正在使用具有非阻塞重试的SpringKafka2.8.0。我使用的是具有固定回退和死信主题的单个重试主题。这是我的配置：

@Bean
public RetryTopicConfiguration receiptRetryConfiguration(
        KafkaTemplate<String, byte[]> kafkaTemplate,
        ConcurrentKafkaListenerContainerFactory<String, byte[]> retryListenerContainerFactory
) {
    return RetryTopicConfigurationBuilder.newInstance()
            .maxAttempts(3)
            .fixedBackOff(5000)
            .setTopicSuffixingStrategy(SUFFIX_WITH_INDEX_VALUE)
            .useSingleTopicForFixedDelays()
            .doNotAutoCreateRetryTopics()
            .dltHandlerMethod(
                    new EndpointHandlerMethod(ConsumerErrorsHandler.class, "handleError")
            )
            .notRetryOn(List.of(SerializationException.class, EmptyMessageValueException.class))
            .listenerFactory(retryListenerContainerFactory)
            .create(kafkaTemplate);
}

我正在尝试实现失败标头管理，并将自定义error_code标头添加到发送到 DLT 的邮件中：

@Bean(RetryTopicInternalBeanNames.DEAD_LETTER_PUBLISHING_RECOVERER_FACTORY_BEAN_NAME)
public DeadLetterPublishingRecovererFactory factory(DestinationTopicResolver resolver) {
    DeadLetterPublishingRecovererFactory factory = new DeadLetterPublishingRecovererFactory(resolver);
    factory.setDeadLetterPublishingRecovererCustomizer(dlpr -> {
        dlpr.setHeadersFunction((record, exception) ->
                new RecordHeaders().add("error_code", getErrorCode(exception).getBytes(StandardCharsets.UTF_8))
        );
    });
    return factory;
}

它可以工作，但调用dlpr.setHeadersFunction（…）将禁用添加重试标头的标准机制：retry_topic-timests、 、 retry_topic-original-timstamp

除了retry_topic-*标头之外，还有什么方法可以将自定义标头添加到发送到DLT的消息中吗？

是否可以将死信发布恢复器工厂配置为仅针对发送到 DLT 主题的消息，而不是为重试主题配置？




Answer 1

请打开新功能问题；我认为我们需要添加一个＜code＞getHeadersFunction（）＜/code＞，这样如果复合函数已经存在，就可以构建它（或者我们应该添加一个 方法）。<="">

https://github.com/spring-projects/spring-kafka/issues

现在，您必须使用反射（例如直接字段访问器）来获取对现有函数（标头函数字段）的引用。