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

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

@@ -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;
@@ -48,6 +49,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Soby Chacko
+ * @author Sanghyoek An
  *
  * @since 1.3
  *
@@ -324,12 +326,37 @@ else if (!(headerName.equals(JSON_TYPES)) && matchesForInbound(headerName)) {
 					populateJsonValueHeader(header, requestedType, headers);
 				}
 				else {
-					headers.put(headerName, headerValueToAddIn(header));
+					handleHeader(headerName, header, headers);
 				}
 			}
 		});
 	}
 
+	/**
+	 * Handle non-reserved headers in {@link DefaultKafkaHeaderMapper}.
+	 * @param headerName the header name.
+	 * @param header the header instance.
+	 * @param headers the target headers.
+	 * @since 4.0.0
+	 */
+	protected void handleHeader(String headerName, Header header, final Map<String, Object> headers) {
+		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) {
 		Class<?> type = Object.class;
 		boolean trusted = false;

spring-kafka/src/main/java/org/springframework/kafka/support/SimpleKafkaHeaderMapper.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));
+						}
+					}
 				}
 			}
 		});