Add possibility to remove custom headers in method: addHeadersFunction(...) from DeadLetterPublishingRecoverer

[wojciechcwek]

Expected Behavior
I'd like to remove current headers in addHeadersFunction(BiFunction<ConsumerRecord, Exception, Headers> headersFunction).

In addHeadersFunction(...) we have consumer record. Removing the header from consumer record has no effect, because previous headers are kept in accept(...) method in headers variable.

Current Behavior

Currently it's not impossible to manipulate existing headers in addHeaderFunction(...)

Context

I want to do delete the current custom header, because headers are passed between all retry topics till dlt topic, which means that when I have 3 retry attempts, then in retry-0 I have one custom header, in retry-1 I have two same custom headers and finally in dlt I will end up with three same custom headers. When there would be an option to delete current headers in that function, then I could do a check if the custom header is already there and if it is, I could delete it before I add it with the new value.

This happens, because Kafka allows to hold duplicate Headers.

My example:

    @Bean(RetryTopicInternalBeanNames.DEAD_LETTER_PUBLISHING_RECOVERER_FACTORY_BEAN_NAME)
    public DeadLetterPublishingRecovererFactory factory(DestinationTopicResolver destinationTopicResolver) {
        DeadLetterPublishingRecovererFactory factory = new DeadLetterPublishingRecovererFactory(destinationTopicResolver);
        factory.setDeadLetterPublishingRecovererCustomizer(dlpr -> dlpr.addHeadersFunction((consumerRecord, exception) -> {
            Throwable rootCause = ExceptionUtils.getRootCause(exception);

            return new RecordHeaders().add(EXTERNAL_SYSTEM_EXCEPTION_MESSAGE_HEADER, rootCause.getMessage().getBytes(StandardCharsets.UTF_8));
        ));
        return factory;
    }