GH-3067: Spring Kafka support multiple headers with same key.

[chickenchickenlove]

• Fixes Spring Kafka does not allow multiple headers with same key #3067

I know that there is an existing PR(#3101) related to this issue, but it seems to have been suspended for long time.
Therefore, I created a new PR to address it.
If you are planning to maintain the existing one, I am happy to close mine.

I approached the problem differently from the existing PR and proposed an alternative solution.
For backward compatibility, user can configure MultiValueKafkaHeaderMapper to KafkaListenerContainerFactory.
For example,

ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
					new ConcurrentKafkaListenerContainerFactory<>();

MultiValueKafkaHeaderMapper headerMapper = new MultiValueKafkaHeaderMapper();

// Create new header mapper.
headerMapper.addSingleValueHeader(SINGLE_VALUE_HEADER);
MessagingMessageConverter converter = new MessagingMessageConverter(headerMapper);

// Set header mapper.
factory.setConsumerFactory(consumerFactory());

[sobychacko]

@chickenchickenlove We were planning to resuscitate the other PR before the next milestone (4.0.0-M3) along with some other refactorings, so please stay tuned for that. Then we can compare your approach in this PR, but I will try first to rebase that PR on top of the latest main and see how it goes.

[chickenchickenlove]

@sobychacko Thanks for your time!!
I didn't get your point clearly.
From what I understand, this PR is no longer needed since you will be taking on a new task.
Could you please confirm if my understanding is correct?

[sobychacko]

Yes, we will need to start from the other PR first. But we will keep this open to compare.

[artembilan]

I don't think that I agree about extra abstraction on the matter.
I think the existing HeaderMapper must support iterables for end-user headers.
So, we should expect some pattern for those custom headers which we can treat as iterable.
And I think that one should be on the in (from Kafka) only.
If we produce and some header is a collection, then populate it into Kafka headers as is.
On the other side, for convenience, we still should support first value mapping by default.
And map only those to collection values, which we have opted-in.

Make sense?

[chickenchickenlove]

@artembilan
There are parts that I understand and parts that I don't.
It's been a while since I last worked with Kafka in my work, so I might be misunderstanding some aspects.
Please keep that in mind as you read this.

How can Spring Kafka anticipate a custom header pattern that is iterable?
Even if Spring Kafka expects a custom header to be iterable, users might always expect the header to be a single value.
What I mean is, users may have already written their code accordingly, like @Header("blahblah") String value.

To handle such cases, I think it would be necessary to add a public method that allows users to configure the pattern for iterable custom headers as a workaround.

If the headers that should be treated as iterable are managed with a whitelist approach,
I believe DefaultKafkaHeaderMapper could support iterable headers without requiring additional abstraction as you mentioned before.

I believe that by moving part of the logic from the MultiValueKafkaHeaderMapper class which I implement to DefaultKafkaHeaderMapper and simply adding a whitelist field,
spring kafka can support multi-header values while maintaining backward compatibility.

Overall, it makes sense to me 👍

[artembilan]

I believe the @Header("blahblah") String value should work. That annotation processor is able to extract single value from an iterable header.

Yes, I would expect some configuration property on the DefaultKafkaHeaderMapper to opt-in to those Kafka headers which have to be mapped to Message header as List.
We cannot use white and black words though. So, we need to come up with other name. Like headerAsListPattern. Or even vararg String... headerAsListPatterns. Where pattern has to follow org.springframework.util.PatternMatchUtils expectations.

On the mapping from Message to Kafka there is no need to think about pattern.
The trigger is simple: if Message is an iterable, then we put it into multi-header in Kafka record.