Addressing PR review

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

Author: chickenchickenlove

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

@@ -182,24 +182,58 @@ If using Spring Boot, it will auto configure this converter bean into the auto-c
 [[multi-value-header]]
 == Support multi-value header
 
-Spring for Apache Kafka 4.0 supports multi-value header.
-To use multi-value header, the patterns for multi-value should be configured to `HeaderMapper`.
-If no patterns are configured to `HeaderMapper`, iterable header values under same header name will be handled like single value.
+Spring for Apache Kafka 4.0 supports multi-value header where same logical header key appears more than once in a Kafka record.
 
-If there are use cases in which multi-value headers are required, use this:
+By default, when multiple headers share the same name, the `HeaderMapper` treats them as a single value and serialises the collection to JSON.
+
+* **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]
 ----
-@Bean
-public void KafkaListenerContainerFactory<Integer, String> containerFactory() {
-    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
-					new ConcurrentKafkaListenerContainerFactory<>();
+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-value*");
+----
+
+Any header whose name matches one of the supplied patterns is
 
-    DefaultKafkaHeaderMapper headerMapper = new DefaultKafkaHeaderMapper();
-    headerMapper.setMultiValueHeaderPatterns("test-multi-value1", "test-multi-value2");
+* **Producer side:** written as separate Kafka headers, one per element.
+* **Consumer side:** reconstructed as a `List<?>` and returned unchanged to the application.
 
-    MessagingMessageConverter converter = new MessagingMessageConverter(headerMapper);
-    factory.setRecordMessageConverter(converter);
-    return factory;
+[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]
+====
+All elements collected under the same multi‑value header **must be of the same Java type**.
+Mixing, for example, `String` and `byte[]` values under a single header key is not supported and will lead to a conversion error.
+====
+
+The mapper is used by multiple components (`KafkaTemplate`, `MessagingMessageConverter`, lister container factories, etc.).
+Define and expose it as a Spring bean wherever it makes sense for your application:
+
+[source,java]
+----
+@Bean
+public DefaultKafkaHeaderMapper multiValueHeaderMapper() {
+    DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
+    mapper.setMultiValueHeaderPatterns("test-multi-*");
+    return mapper;
 }
 ----

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

@@ -75,4 +75,4 @@ For details, see xref:kafka/receiving-messages/rebalance-listeners.adoc#new-reba
 === Support multi-value header
 
 Spring for Apache Kafka 4.0 supports multi-value header for Kafka Record.
-For details, see xref:kafka/headers.adoc#multi-value-header[Support multi-value header].
+More details are available in xref:kafka/headers.adoc#multi-value-header[Support multi-value header].

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

@@ -296,7 +296,6 @@ protected void fromUserHeader(String headerName, Header header, final Map<String
 			List<Object> headerValues = (List<Object>)
 					headers.computeIfAbsent(headerName, key -> new ArrayList<>());
 			headerValues.add(headerValueToAddIn(header));
-//			headers.put(headerName, headerValues);
 		}
 	}
 

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

@@ -333,12 +333,14 @@ private void fromHeader(String key, Object rawValue, Map<String, String> jsonHea
 					className = JAVA_LANG_STRING;
 					encodeToJson = true;
 				}
+				final byte[] calculatedValue;
 				if (!encodeToJson && valueToAdd instanceof String) {
-					target.add(new RecordHeader(key, ((String) valueToAdd).getBytes(getCharset())));
+					calculatedValue = ((String) valueToAdd).getBytes(getCharset());
 				}
 				else {
-					target.add(new RecordHeader(key, headerObjectMapper.writeValueAsBytes(valueToAdd)));
+					calculatedValue = headerObjectMapper.writeValueAsBytes(valueToAdd);
 				}
+				target.add(new RecordHeader(key, calculatedValue));
 				jsonHeaders.putIfAbsent(key, className);
 			}
 			catch (Exception e) {

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

@@ -126,9 +126,8 @@ public void toHeaders(Headers source, Map<String, Object> headers) {
 	}
 
 	private void fromHeader(String key, Object value, Headers target) {
-		Object valueToAdd = headerValueToAddOut(key, value);
-		if (valueToAdd instanceof byte[] && matches(key, valueToAdd)) {
-			target.add(new RecordHeader(key, (byte[]) valueToAdd));
+		if (headerValueToAddOut(key, value) instanceof byte[] valueToAdd && matches(key, valueToAdd)) {
+			target.add(new RecordHeader(key, valueToAdd));
 		}
 	}
 

spring-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java

@@ -339,10 +339,12 @@ void multiValueHeaderToTest() {
 		// GIVEN
 		String multiValueHeader1 = "test-multi-value1";
 		String multiValueHeader2 = "test-multi-value2";
+		String multiValueWildeCardHeader1 = "test-wildcard-value1";
+		String multiValueWildeCardHeader2 = "test-wildcard-value2";
 		String singleValueHeader = "test-single-value1";
 
 		DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
-		mapper.setMultiValueHeaderPatterns(multiValueHeader1, multiValueHeader2);
+		mapper.setMultiValueHeaderPatterns(multiValueHeader1, multiValueHeader2, "test-wildcard-*");
 
 		Headers rawHeaders = new RecordHeaders();
 		rawHeaders.add(KafkaHeaders.DELIVERY_ATTEMPT, new byte[] { 0, 0, 0, 1 });
@@ -358,6 +360,11 @@ void multiValueHeaderToTest() {
 		rawHeaders.add(multiValueHeader2, new byte[] { 0, 0, 0, 4 });
 		rawHeaders.add(multiValueHeader2, new byte[] { 0, 0, 0, 5 });
 
+		rawHeaders.add(multiValueWildeCardHeader1, new byte[] { 0, 0, 0, 6 });
+		rawHeaders.add(multiValueWildeCardHeader1, new byte[] { 0, 0, 0, 7 });
+		rawHeaders.add(multiValueWildeCardHeader2, new byte[] { 0, 0, 0, 8 });
+		rawHeaders.add(multiValueWildeCardHeader2, new byte[] { 0, 0, 0, 9 });
+
 		// WHEN
 		Map<String, Object> mappedHeaders = new HashMap<>();
 		mapper.toHeaders(rawHeaders, mappedHeaders);
@@ -379,13 +386,25 @@ void multiValueHeaderToTest() {
 		List<byte[]> multiHeader2Values = (List<byte[]>) mappedHeaders.get(multiValueHeader2);
 		assertThat(multiHeader2Values).contains(new byte[] { 0, 0, 0, 4 },
 												new byte[] { 0, 0, 0, 5 });
+
+		@SuppressWarnings("unchecked")
+		List<byte[]> multiValueWildeCardHeader1Values = (List<byte[]>) mappedHeaders.get(multiValueWildeCardHeader1);
+		assertThat(multiValueWildeCardHeader1Values).contains(new byte[] { 0, 0, 0, 6 },
+															new byte[] { 0, 0, 0, 7 });
+
+		@SuppressWarnings("unchecked")
+		List<byte[]> multiValueWildeCardHeader2Values = (List<byte[]>) mappedHeaders.get(multiValueWildeCardHeader2);
+		assertThat(multiValueWildeCardHeader2Values).contains(new byte[] { 0, 0, 0, 8 },
+															new byte[] { 0, 0, 0, 9 });
 	}
 
 	@Test
 	void multiValueHeaderFromTest() {
 		// GIVEN
 		String multiValueHeader1 = "test-multi-value1";
 		String multiValueHeader2 = "test-multi-value2";
+		String multiValueHeader3 = "test-other-multi-value1";
+		String multiValueHeader4 = "test-prefix-match-multi";
 		String singleValueHeader = "test-single-value1";
 
 		Message<String> message = MessageBuilder
@@ -394,11 +413,17 @@ void multiValueHeaderFromTest() {
 													new byte[] { 0, 0, 0, 2 }))
 				.setHeader(multiValueHeader2, List.of(new byte[] { 0, 0, 0, 3 },
 													new byte[] { 0, 0, 0, 4 }))
+				.setHeader(multiValueHeader3, List.of(new byte[] { 0, 0, 0, 9 },
+													new byte[] { 0, 0, 0, 10 }))
+				.setHeader(multiValueHeader4, List.of(new byte[] { 0, 0, 0, 11 },
+													new byte[] { 0, 0, 0, 12 }))
 				.setHeader(singleValueHeader, new byte[] { 0, 0, 0, 5 })
 				.build();
 
 		DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
-		mapper.setMultiValueHeaderPatterns(multiValueHeader1, multiValueHeader2);
+		mapper.setMultiValueHeaderPatterns("test-multi-*",
+											multiValueHeader3,
+											"*-prefix-match-multi*");
 
 		// WHEN
 		Headers results = new RecordHeaders();
@@ -410,6 +435,10 @@ void multiValueHeaderFromTest() {
 				new RecordHeader(multiValueHeader1, new byte[] { 0, 0, 0, 2 }),
 				new RecordHeader(multiValueHeader2, new byte[] { 0, 0, 0, 3 }),
 				new RecordHeader(multiValueHeader2, new byte[] { 0, 0, 0, 4 }),
+				new RecordHeader(multiValueHeader3, new byte[] { 0, 0, 0, 9 }),
+				new RecordHeader(multiValueHeader3, new byte[] { 0, 0, 0, 10 }),
+				new RecordHeader(multiValueHeader4, new byte[] { 0, 0, 0, 11 }),
+				new RecordHeader(multiValueHeader4, new byte[] { 0, 0, 0, 12 }),
 				new RecordHeader(singleValueHeader, new byte[] { 0, 0, 0, 5 })
 		);
 	}