Addressing PR review

Signed-off-by: chickenchickenlove <ojt90902@naver.com>

Author: chickenchickenlove

spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java

@@ -65,6 +65,8 @@ public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper {
 
 	private final List<HeaderMatcher> matchers = new ArrayList<>();
 
+	private final List<HeaderMatcher> headerMatchersForMultiValue = new ArrayList<>();
+
 	private final Map<String, Boolean> rawMappedHeaders = new HashMap<>();
 
 	{
@@ -191,6 +193,16 @@ public void addRawMappedHeader(String name, boolean toString) {
 		this.rawMappedHeaders.put(name, toString);
 	}
 
+	/**
+	 * Add patterns for matching multi-value headers under the same key.
+	 * @param patterns the patterns for header.
+	 */
+	public void addHeaderPatternsForMultiValue(String ... patterns) {
+		for (String pattern : patterns) {
+			this.headerMatchersForMultiValue.add(new SimplePatternBasedHeaderMatcher(pattern));
+		}
+	}
+
 	protected boolean matches(String header, Object value) {
 		if (matches(header)) {
 			if ((header.equals(MessageHeaders.REPLY_CHANNEL) || header.equals(MessageHeaders.ERROR_CHANNEL))
@@ -251,6 +263,20 @@ protected Object headerValueToAddOut(String key, Object value) {
 		return valueToAdd;
 	}
 
+	/**
+	 * Check whether the header value should be mapped to multiple values.
+	 * @param headerName the header name.
+	 * @return True for multiple values at the same key.
+	 */
+	protected boolean doesMatchMultiValueHeader(String headerName) {
+		for (HeaderMatcher headerMatcher : this.headerMatchersForMultiValue) {
+			if (headerMatcher.matchHeader(headerName)) {
+				return true;
+			}
+		}
+		return false;
+	}
+
 	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	@Nullable
 	private byte[] mapRawOut(String header, Object value) {

spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 the original author or authors.
+ * Copyright 2017-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -338,9 +339,22 @@ else if (!(headerName.equals(JSON_TYPES)) && matchesForInbound(headerName)) {
 	 * @param headers the target headers.
 	 * @since 4.0.0
 	 */
-
 	protected void handleHeader(String headerName, Header header, final Map<String, Object> headers) {
-		headers.put(headerName, headerValueToAddIn(header));
+		if (!this.doesMatchMultiValueHeader(headerName)) {
+			headers.put(headerName, headerValueToAddIn(header));
+		}
+		else {
+			Object values = headers.getOrDefault(headerName, new ArrayList<>());
+			if (values instanceof List) {
+				@SuppressWarnings("unchecked")
+				List<Object> castedValues = (List<Object>) values;
+				castedValues.add(headerValueToAddIn(header));
+				headers.put(headerName, castedValues);
+			}
+			else {
+				headers.put(headerName, headerValueToAddIn(header));
+			}
+		}
 	}
 
 	private void populateJsonValueHeader(Header header, String requestedType, Map<String, Object> headers) {

spring-kafka/src/main/java/org/springframework/kafka/support/MultiValueKafkaHeaderMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2022 the original author or authors.
+ * Copyright 2018-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,7 +17,9 @@
 package org.springframework.kafka.support;
 
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -35,6 +37,7 @@
  * The exceptions are correlation and reply headers for request/reply
  *
  * @author Gary Russell
+ * @author Sanghyeok An
  * @since 2.1.3
  *
  */
@@ -111,7 +114,21 @@ public void toHeaders(Headers source, Map<String, Object> target) {
 					target.put(headerName, ByteBuffer.wrap(header.value()).getInt());
 				}
 				else {
-					target.put(headerName, headerValueToAddIn(header));
+					if (!this.doesMatchMultiValueHeader(headerName)) {
+						target.put(headerName, headerValueToAddIn(header));
+					}
+					else {
+						Object values = target.getOrDefault(headerName, new ArrayList<>());
+						if (values instanceof List) {
+							@SuppressWarnings("unchecked")
+							List<Object> castedValues = (List<Object>) values;
+							castedValues.add(headerValueToAddIn(header));
+							target.put(headerName, castedValues);
+						}
+						else {
+							target.put(headerName, headerValueToAddIn(header));
+						}
+					}
 				}
 			}
 		});

spring-kafka/src/main/java/org/springframework/kafka/support/SimpleKafkaHeaderMapper.java

@@ -58,7 +58,6 @@
  * @author Gary Russell
  * @author Dariusz Szablinski
  * @author Biju Kunjummen
- * @author Sanghyeok An
  */
 public class MessagingMessageConverter implements RecordMessageConverter {
 
@@ -84,15 +83,6 @@ public MessagingMessageConverter() {
 		this(msg -> msg.getHeaders().get(KafkaHeaders.PARTITION, Integer.class));
 	}
 
-	/**
-	 * Construct an instance that uses given HeaderMapper.
-	 * @param headerMapper the Header mapper.
-	 * @since 4.0.0
-	 */
-	public MessagingMessageConverter(KafkaHeaderMapper headerMapper) {
-		this(msg -> msg.getHeaders().get(KafkaHeaders.PARTITION, Integer.class), headerMapper);
-	}
-
 	/**
 	 * Construct an instance that uses the supplied partition provider function. The
 	 * function can return null to delegate the partition selection to the kafka client.
@@ -110,18 +100,6 @@ public MessagingMessageConverter(Function<Message<?>, @Nullable Integer> partiti
 		this.partitionProvider = partitionProvider;
 	}
 
-	/**
-	 * Construct an instance that uses the supplied partition provider function and given HeaderMapper.
-	 * @param partitionProvider the provider.
-	 * @param headerMapper the Header mapper.
-	 * @since 4.0.0
-	 */
-	public MessagingMessageConverter(Function<Message<?>, @Nullable Integer> partitionProvider, KafkaHeaderMapper headerMapper) {
-		Assert.notNull(partitionProvider, "'partitionProvider' cannot be null");
-		this.headerMapper = headerMapper;
-		this.partitionProvider = partitionProvider;
-	}
-
 	/**
 	 * Generate {@link Message} {@code ids} for produced messages. If set to {@code false},
 	 * will try to use a default value. By default set to {@code false}.