GH-3067: support multi-value headers

Fixes: #3067

* Add java docs and test case for SimpleKafkaHeaderMapper.

* Add docs for new feature

Signed-off-by: Sanghyeok An <[email protected]>

Author: chickenchickenlove

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/headers.adoc

@@ -179,3 +179,40 @@ MessagingMessageConverter converter() {
 
 If using Spring Boot, it will auto configure this converter bean into the auto-configured `KafkaTemplate`; otherwise you should add this converter to the template.
 
+[[multi-value-header]]
+== Support multi-value header mapping
+
+Starting with 4.0, multi-value header mapping is supported, where the same logical header key appears more than once in a Kafka record.
+
+By default, the `HeaderMapper` does **not** create multiple Kafka headers with the same name.
+Instead, when it encounters a collection value (e.g., a `List<byte[]>`), it serializes the entire collection into **one** Kafka header whose value is a JSON array.
+
+* **Producer side:** `DefaultKafkaHeaderMapper` writes the JSON bytes, while `SimpleKafkaHeaderMapper` ignore it.
+* **Consumer side:** the mapper exposes the header as a single value—the **last occurrence wins**; earlier duplicates are silently discarded.
+
+Preserving each individual header requires explicit registration of patterns that designate the header as multi‑valued.
+
+`DefaultKafkaHeaderMapper#setMultiValueHeaderPatterns(String... patterns)` accepts a list of patterns, which can be either wildcard expressions or exact header names.
+
+[source, java]
+----
+DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
+
+// Explicit header names
+mapper.setMultiValueHeaderPatterns("test-multi-value1", "test-multi-value2");
+
+// Wildcard patterns for test-multi-value1, test-multi-value2
+mapper.setMultiValueHeaderPatterns("test-multi-*");
+----
+
+Any header whose name matches one of the supplied patterns is
+
+* **Producer side:** written as separate Kafka headers, one per element.
+* **Consumer side:** collected into a `List<?>` that contains the individual header values; each element is returned to the application **after the usual deserialization or type conversion performed by the configured `HeaderMapper`.**
+
+NOTE: Regular expressions are *not* supported; only the +*+ wildcard is allowed in simple patterns—supporting direct equality and forms such as: +xxx*+, +*xxx+, +*xxx*+, +xxx*yyy+.
+
+[IMPORTANT]
+====
+On the *Producer Side*, When `DefaultKafkaHeaderMapper` serializes a multi-value header, every element in that collection must be of a single Java type—mixing, for example, `String` and `byte[]` values under a single header key will lead to a conversion error.
+====

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

@@ -70,3 +70,9 @@ Several deprecated items have been removed:
 
 Spring for Apache Kafka 4.0 supports Kafka 4.0’s new consumer rebalance protocol - https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol[KIP-848].
 For details, see xref:kafka/receiving-messages/rebalance-listeners.adoc#new-rebalalcne-protocol[New Consumer Rebalace Protocol docs].
+
+[[x40-multi-value-header]]
+=== Support multi-value header
+
+The `DefaultKafkaHeaderMapper` and `SimpleKafkaHeaderMapper` support multi-value header mapping for Kafka records.
+More details are available in xref:kafka/headers.adoc#multi-value-header[Support multi-value header mapping].

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> multiValueHeaderMatchers = new ArrayList<>();
+
 	private final Map<String, Boolean> rawMappedHeaders = new HashMap<>();
 
 	{
@@ -191,6 +193,18 @@ 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.
+	 * @since 4.0
+	 */
+	public void setMultiValueHeaderPatterns(String ... patterns) {
+		this.multiValueHeaderMatchers.addAll(Arrays
+												.stream(patterns)
+												.map(SimplePatternBasedHeaderMatcher::new)
+												.toList());
+	}
+
 	protected boolean matches(String header, Object value) {
 		if (matches(header)) {
 			if ((header.equals(MessageHeaders.REPLY_CHANNEL) || header.equals(MessageHeaders.ERROR_CHANNEL))
@@ -251,6 +265,40 @@ 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.
+	 * @since 4.0
+	 */
+	protected boolean doesMatchMultiValueHeader(String headerName) {
+		for (HeaderMatcher headerMatcher : this.multiValueHeaderMatchers) {
+			if (headerMatcher.matchHeader(headerName)) {
+				return true;
+			}
+		}
+		return false;
+	}
+
+	/**
+	 * 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
+	 */
+	protected void fromUserHeader(String headerName, Header header, final Map<String, Object> headers) {
+		if (!doesMatchMultiValueHeader(headerName)) {
+			headers.put(headerName, headerValueToAddIn(header));
+		}
+		else {
+			@SuppressWarnings("unchecked")
+			List<Object> headerValues = (List<Object>)
+					headers.computeIfAbsent(headerName, key -> new ArrayList<>());
+			headerValues.add(headerValueToAddIn(header));
+		}
+	}
+
 	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	@Nullable
 	private byte[] mapRawOut(String header, Object value) {

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

@@ -48,6 +48,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Soby Chacko
+ * @author Sanghyoek An
  *
  * @since 1.3
  *
@@ -266,31 +267,17 @@ public void fromHeaders(MessageHeaders headers, Headers target) {
 		final ObjectMapper headerObjectMapper = getObjectMapper();
 		headers.forEach((key, rawValue) -> {
 			if (matches(key, rawValue)) {
-				Object valueToAdd = headerValueToAddOut(key, rawValue);
-				if (valueToAdd instanceof byte[]) {
-					target.add(new RecordHeader(key, (byte[]) valueToAdd));
-				}
-				else {
-					try {
-						String className = valueToAdd.getClass().getName();
-						boolean encodeToJson = this.encodeStrings;
-						if (this.toStringClasses.contains(className)) {
-							valueToAdd = valueToAdd.toString();
-							className = JAVA_LANG_STRING;
-							encodeToJson = true;
-						}
-						if (!encodeToJson && valueToAdd instanceof String) {
-							target.add(new RecordHeader(key, ((String) valueToAdd).getBytes(getCharset())));
-						}
-						else {
-							target.add(new RecordHeader(key, headerObjectMapper.writeValueAsBytes(valueToAdd)));
-						}
-						jsonHeaders.put(key, className);
+				if (doesMatchMultiValueHeader(key)) {
+					if (rawValue instanceof Iterable<?> valuesToMap) {
+						valuesToMap.forEach(o -> fromHeader(key, o, jsonHeaders, headerObjectMapper, target));
 					}
-					catch (Exception e) {
-						logger.error(e, () -> "Could not map " + key + " with type " + rawValue.getClass().getName());
+					else {
+						fromHeader(key, rawValue, jsonHeaders, headerObjectMapper, target);
 					}
 				}
+				else {
+					fromHeader(key, rawValue, jsonHeaders, headerObjectMapper, target);
+				}
 			}
 		});
 		if (!jsonHeaders.isEmpty()) {
@@ -324,12 +311,44 @@ else if (!(headerName.equals(JSON_TYPES)) && matchesForInbound(headerName)) {
 					populateJsonValueHeader(header, requestedType, headers);
 				}
 				else {
-					headers.put(headerName, headerValueToAddIn(header));
+					fromUserHeader(headerName, header, headers);
 				}
 			}
 		});
 	}
 
+	private void fromHeader(String key, Object rawValue, Map<String, String> jsonHeaders,
+							ObjectMapper headerObjectMapper, Headers target) {
+
+		Object valueToAdd = headerValueToAddOut(key, rawValue);
+		if (valueToAdd instanceof byte[]) {
+			target.add(new RecordHeader(key, (byte[]) valueToAdd));
+		}
+		else {
+			try {
+				String className = valueToAdd.getClass().getName();
+				boolean encodeToJson = this.encodeStrings;
+				if (this.toStringClasses.contains(className)) {
+					valueToAdd = valueToAdd.toString();
+					className = JAVA_LANG_STRING;
+					encodeToJson = true;
+				}
+				final byte[] calculatedValue;
+				if (!encodeToJson && valueToAdd instanceof String) {
+					calculatedValue = ((String) valueToAdd).getBytes(getCharset());
+				}
+				else {
+					calculatedValue = headerObjectMapper.writeValueAsBytes(valueToAdd);
+				}
+				target.add(new RecordHeader(key, calculatedValue));
+				jsonHeaders.putIfAbsent(key, className);
+			}
+			catch (Exception e) {
+				logger.error(e, () -> "Could not map " + key + " with type " + rawValue.getClass().getName());
+			}
+		}
+	}
+
 	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.
@@ -35,6 +35,7 @@
  * The exceptions are correlation and reply headers for request/reply
  *
  * @author Gary Russell
+ * @author Sanghyeok An
  * @since 2.1.3
  *
  */
@@ -94,27 +95,40 @@ public static SimpleKafkaHeaderMapper forInboundOnlyWithMatchers(String... patte
 	public void fromHeaders(MessageHeaders headers, Headers target) {
 		headers.forEach((key, value) -> {
 			if (!NEVER.contains(key)) {
-				Object valueToAdd = headerValueToAddOut(key, value);
-				if (valueToAdd instanceof byte[] && matches(key, valueToAdd)) {
-					target.add(new RecordHeader(key, (byte[]) valueToAdd));
+				if (doesMatchMultiValueHeader(key)) {
+					if (value instanceof Iterable<?> valuesToMap) {
+						valuesToMap.forEach(o -> fromHeader(key, o, target));
+					}
+					else {
+						fromHeader(key, value, target);
+					}
+				}
+				else {
+					fromHeader(key, value, target);
 				}
 			}
 		});
 	}
 
 	@Override
-	public void toHeaders(Headers source, Map<String, Object> target) {
+	public void toHeaders(Headers source, Map<String, Object> headers) {
 		source.forEach(header -> {
 			String headerName = header.key();
 			if (matchesForInbound(headerName)) {
 				if (headerName.equals(KafkaHeaders.DELIVERY_ATTEMPT)) {
-					target.put(headerName, ByteBuffer.wrap(header.value()).getInt());
+					headers.put(headerName, ByteBuffer.wrap(header.value()).getInt());
 				}
 				else {
-					target.put(headerName, headerValueToAddIn(header));
+					fromUserHeader(headerName, header, headers);
 				}
 			}
 		});
 	}
 
+	private void fromHeader(String key, Object value, Headers target) {
+		if (headerValueToAddOut(key, value) instanceof byte[] valueToAdd && matches(key, valueToAdd)) {
+			target.add(new RecordHeader(key, valueToAdd));
+		}
+	}
+
 }