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

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

@@ -23,6 +23,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -79,6 +80,12 @@ public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper {
 
 	private Charset charset = StandardCharsets.UTF_8;
 
+	private final List<HeaderMatcher> matchersForListValue = new ArrayList<>();
+
+	private final Set<String> cachedHeadersForListValue = new LinkedHashSet<>();
+
+	private final Set<String> cachedHeadersForSingleValue = new LinkedHashSet<>();
+
 	/**
 	 * Construct a mapper that will match the supplied patterns (outbound) and all headers
 	 * (inbound). For outbound mapping, certain internal framework headers are never
@@ -97,6 +104,20 @@ public AbstractKafkaHeaderMapper(String... patterns) {
 	 * @param patterns the patterns.
 	 */
 	protected AbstractKafkaHeaderMapper(boolean outbound, String... patterns) {
+		this(outbound, new ArrayList<>(), patterns);
+	}
+
+	/**
+	 * Construct a mapper that will match the supplied patterns (outbound) and all headers
+	 * (inbound). For outbound mapping, certain internal framework headers are never
+	 * mapped. For inbound mapping, Headers that match the pattern specified in
+	 * {@code patternsForListValue} will be appended to the values under the same key.
+	 *
+	 * @param outbound true for an outbound mapper.
+	 * @param patternsForListValue the patterns for multiple values at the same key.
+	 * @param patterns the patterns.
+	 */
+	protected AbstractKafkaHeaderMapper(boolean outbound, List<String> patternsForListValue, String... patterns) {
 		Assert.notNull(patterns, "'patterns' must not be null");
 		this.outbound = outbound;
 		if (outbound) {
@@ -123,6 +144,11 @@ protected AbstractKafkaHeaderMapper(boolean outbound, String... patterns) {
 		for (String pattern : patterns) {
 			this.matchers.add(new SimplePatternBasedHeaderMatcher(pattern));
 		}
+
+		for (String patternForListValue : patternsForListValue) {
+			this.matchersForListValue.add(new SimplePatternBasedHeaderMatcher(patternForListValue));
+		}
+
 	}
 
 	/**
@@ -287,6 +313,34 @@ private String mapRawIn(String header, byte[] value) {
 		return null;
 	}
 
+	/**
+	 * 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 isHeaderForListValue(String headerName) {
+		if (this.matchersForListValue.isEmpty()) {
+			return false;
+		}
+
+		if (this.cachedHeadersForSingleValue.contains(headerName)) {
+			return false;
+		}
+
+		if (this.cachedHeadersForListValue.contains(headerName)) {
+			return true;
+		}
+
+		for (HeaderMatcher headerMatcher : this.matchersForListValue) {
+			if (headerMatcher.matchHeader(headerName)) {
+				this.cachedHeadersForListValue.add(headerName);
+				return true;
+			}
+		}
+		this.cachedHeadersForSingleValue.add(headerName);
+		return false;
+	}
+
 	/**
 	 * A matcher for headers.
 	 * @since 2.3

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 Sanghyeok An
  *
  * @since 1.3
  *
@@ -102,6 +104,27 @@ public DefaultKafkaHeaderMapper() {
 		this(JacksonUtils.enhancedObjectMapper());
 	}
 
+	/**
+	 * Construct an instance with the default object mapper and default header patterns
+	 * for outbound headers and default header patterns for inbound multi-value headers;
+	 * all inbound headers are mapped. The default pattern list is
+	 * {@code "!id", "!timestamp" and "*"}. In addition, most of the headers in
+	 * {@link KafkaHeaders} are never mapped as headers since they represent data in
+	 * consumer/producer records.
+	 * Headers that match the pattern specified in {@code patternsForListValue} will be
+	 * appended to the values under the same key.
+	 *
+	 * @param patternsForListValue the patterns for multiple values at the same key.
+	 * @see #DefaultKafkaHeaderMapper(ObjectMapper)
+	 */
+	public DefaultKafkaHeaderMapper(List<String> patternsForListValue) {
+		this(JacksonUtils.enhancedObjectMapper(),
+		patternsForListValue,
+		"!" + MessageHeaders.ID,
+		"!" + MessageHeaders.TIMESTAMP,
+		"*");
+	}
+
 	/**
 	 * Construct an instance with the provided object mapper and default header patterns
 	 * for outbound headers; all inbound headers are mapped. The patterns are applied in
@@ -148,11 +171,32 @@ public DefaultKafkaHeaderMapper(String... patterns) {
 	 * @see org.springframework.util.PatternMatchUtils#simpleMatch(String, String)
 	 */
 	public DefaultKafkaHeaderMapper(ObjectMapper objectMapper, String... patterns) {
-		this(true, objectMapper, patterns);
+		this(true, objectMapper, new ArrayList<>(), patterns);
+	}
+
+	/**
+	 * Construct an instance with the provided object mapper and the provided header
+	 * patterns for outbound headers; all inbound headers are mapped. The patterns are
+	 * applied in order, stopping on the first match (positive or negative). Patterns are
+	 * negated by preceding them with "!". The patterns will replace the default patterns;
+	 * you generally should not map the {@code "id" and "timestamp"} headers. Note: most
+	 * of the headers in {@link KafkaHeaders} are never mapped as headers since they
+	 * represent data in consumer/producer records.
+	 * @param objectMapper the object mapper.
+	 * @param patternsForListValue the patterns for multiple values at the same key.
+	 * @param patterns the patterns.
+	 * @see org.springframework.util.PatternMatchUtils#simpleMatch(String, String)
+	 */
+	public DefaultKafkaHeaderMapper(ObjectMapper objectMapper, List<String> patternsForListValue, String... patterns) {
+		this(true, objectMapper, patternsForListValue, patterns);
 	}
 
 	private DefaultKafkaHeaderMapper(boolean outbound, ObjectMapper objectMapper, String... patterns) {
-		super(outbound, patterns);
+		this(outbound, objectMapper, new ArrayList<>(), patterns);
+	}
+
+	private DefaultKafkaHeaderMapper(boolean outbound, ObjectMapper objectMapper, List<String> patternsForListValue, String... patterns) {
+		super(outbound, patternsForListValue, patterns);
 		Assert.notNull(objectMapper, "'objectMapper' must not be null");
 		Assert.noNullElements(patterns, "'patterns' must not have null elements");
 		this.objectMapper = objectMapper;
@@ -324,12 +368,39 @@ 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.isHeaderForListValue(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,8 @@
  * The exceptions are correlation and reply headers for request/reply
  *
  * @author Gary Russell
+ * @author Sanghyeok An
+ *
  * @since 2.1.3
  *
  */
@@ -61,6 +65,25 @@ public SimpleKafkaHeaderMapper() {
 				"*");
 	}
 
+	/**
+	 * Construct an instance with the default object mapper and default header patterns
+	 * for outbound headers default header patterns for inbound multi-value headers;
+	 * all inbound headers are mapped. The default pattern list is
+	 * {@code "!id", "!timestamp" and "*"}. In addition, most of the headers in
+	 * {@link KafkaHeaders} are never mapped as headers since they represent data in
+	 * consumer/producer records.
+	 * Headers that match the pattern specified in {@code patternsForListValue} will be
+	 * appended to the values under the same key.
+	 *
+	 * @param patternsForMultiValue the patterns for multiple values at the same key.
+	 */
+	public SimpleKafkaHeaderMapper(List<String> patternsForMultiValue) {
+		this(true, patternsForMultiValue,
+		"!" + MessageHeaders.ID,
+			"!" + MessageHeaders.TIMESTAMP,
+			"*");
+	}
+
 	/**
 	 * Construct an instance with a default object mapper and the provided header patterns
 	 * for outbound headers; all inbound headers are mapped. The patterns are applied in
@@ -77,7 +100,11 @@ public SimpleKafkaHeaderMapper(String... patterns) {
 	}
 
 	private SimpleKafkaHeaderMapper(boolean outbound, String... patterns) {
-		super(outbound, patterns);
+		this(outbound, new ArrayList<>(), patterns);
+	}
+
+	private SimpleKafkaHeaderMapper(boolean outbound, List<String> patternsForListValue, String... patterns) {
+		super(outbound, patternsForListValue, patterns);
 	}
 
 	/**
@@ -111,7 +138,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.isHeaderForListValue(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/converter/MessagingMessageConverter.java

@@ -58,6 +58,7 @@
  * @author Gary Russell
  * @author Dariusz Szablinski
  * @author Biju Kunjummen
+ * @author Sanghyeok An
  */
 public class MessagingMessageConverter implements RecordMessageConverter {
 
@@ -83,6 +84,15 @@ 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.
@@ -100,6 +110,18 @@ 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}.