diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java index 685fdb3f..f13fd5a1 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-2019 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. @@ -16,16 +16,23 @@ package org.springframework.kafka.support; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; import java.text.MessageFormat; import java.util.ArrayList; import java.util.Arrays; +import java.util.HashMap; import java.util.List; +import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.kafka.common.header.Header; +import org.springframework.lang.Nullable; import org.springframework.messaging.MessageHeaders; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; import org.springframework.util.PatternMatchUtils; /** @@ -58,6 +65,12 @@ public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper { protected final List matchers = new ArrayList<>(NEVER_MAPPED); // NOSONAR + private final Map rawMappedtHeaders = new HashMap<>(); + + private boolean mapAllStringsOut; + + private Charset charset = StandardCharsets.UTF_8; + public AbstractKafkaHeaderMapper(String... patterns) { Assert.notNull(patterns, "'patterns' must not be null"); for (String pattern : patterns) { @@ -65,6 +78,49 @@ public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper { } } + /** + * Set to true to map all {@code String} valued outbound headers to {@code byte[]}. + * To map to a {@code String} for inbound, there must be an entry in the rawMappedHeaders map. + * @param mapAllStringsOut true to map all strings. + * @since 2.2.5 + * @see #setRawMappedHaeaders(Map) + */ + public void setMapAllStringsOut(boolean mapAllStringsOut) { + this.mapAllStringsOut = mapAllStringsOut; + } + + protected Charset getCharset() { + return this.charset; + } + + /** + * Set the charset to use when mapping String-valued headers to/from byte[]. Default UTF-8. + * @param charset the charset. + * @since 2.2.5 + * @see #setRawMappedHaeaders(Map) + */ + public void setCharset(Charset charset) { + Assert.notNull(charset, "'charset' cannot be null"); + this.charset = charset; + } + + /** + * Set the headers to not perform any conversion on (except {@code String} to + * {@code byte[]} for outbound). Inbound headers that match will be mapped as + * {@code byte[]} unless the corresponding boolean in the map value is true, + * in which case it will be mapped as a String. + * @param rawMappedHeaders the header names to not convert and + * @since 2.2.5 + * @see #setCharset(Charset) + * @see #setMapAllStringsOut(boolean) + */ + public void setRawMappedHaeaders(Map rawMappedHeaders) { + if (!ObjectUtils.isEmpty(rawMappedHeaders)) { + this.rawMappedtHeaders.clear(); + this.rawMappedtHeaders.putAll(rawMappedHeaders); + } + } + protected boolean matches(String header, Object value) { if (matches(header)) { if ((header.equals(MessageHeaders.REPLY_CHANNEL) || header.equals(MessageHeaders.ERROR_CHANNEL)) @@ -93,6 +149,57 @@ public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper { return false; } + /** + * Check if the value is a String and convert to byte[], if so configured. + * @param key the header name. + * @param value the headet value. + * @return the value to add. + * @since 2.2.5 + */ + protected Object headerValueToAddOut(String key, Object value) { + Object valueToAdd = mapRawOut(key, value); + if (valueToAdd == null) { + valueToAdd = value; + } + return valueToAdd; + } + + @Nullable + private byte[] mapRawOut(String header, Object value) { + if (this.mapAllStringsOut || this.rawMappedtHeaders.containsKey(header)) { + if (value instanceof byte[]) { + return (byte[]) value; + } + else if (value instanceof String) { + return ((String) value).getBytes(this.charset); + } + } + return null; + } + + /** + * Check if the header value should be mapped to a String, if so configured. + * @param header the header. + * @return the value to add. + */ + protected Object headertValueToAddIn(Header header) { + Object mapped = mapRawIn(header.key(), header.value()); + if (mapped == null) { + mapped = header.value(); + } + return mapped; + } + + @Nullable + private String mapRawIn(String header, byte[] value) { + Boolean asString = this.rawMappedtHeaders.get(header); + if (Boolean.TRUE.equals(asString)) { + return new String(value, this.charset); + } + return null; + } + + /** * A pattern-based header matcher that matches if the specified * header matches the specified simple pattern. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java index 9f746760..195b7cd5 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2018 the original author or authors. + * Copyright 2017-2019 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. @@ -183,11 +183,11 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper { * If any of the supplied packages is {@code "*"}, all packages are trusted. * If a class for a non-trusted package is encountered, the header is returned to the * application with value of type {@link NonTrustedHeaderType}. - * @param trustedPackages the packages to trust. + * @param packagesToTrust the packages to trust. */ - public void addTrustedPackages(String... trustedPackages) { - if (trustedPackages != null) { - for (String whiteList : trustedPackages) { + public void addTrustedPackages(String... packagesToTrust) { + if (packagesToTrust != null) { + for (String whiteList : packagesToTrust) { if ("*".equals(whiteList)) { this.trustedPackages.clear(); break; @@ -213,25 +213,26 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper { public void fromHeaders(MessageHeaders headers, Headers target) { final Map jsonHeaders = new HashMap<>(); final ObjectMapper headerObjectMapper = getObjectMapper(); - headers.forEach((k, v) -> { - if (matches(k, v)) { - if (v instanceof byte[]) { - target.add(new RecordHeader(k, (byte[]) v)); + headers.forEach((key, val) -> { + if (matches(key, val)) { + Object valueToAdd = headerValueToAddOut(key, val); + if (valueToAdd instanceof byte[]) { + target.add(new RecordHeader(key, (byte[]) valueToAdd)); } else { try { - Object value = v; - String className = v.getClass().getName(); + Object value = valueToAdd; + String className = valueToAdd.getClass().getName(); if (this.toStringClasses.contains(className)) { - value = v.toString(); + value = valueToAdd.toString(); className = "java.lang.String"; } - target.add(new RecordHeader(k, headerObjectMapper.writeValueAsBytes(value))); - jsonHeaders.put(k, className); + target.add(new RecordHeader(key, headerObjectMapper.writeValueAsBytes(value))); + jsonHeaders.put(key, className); } - catch (Exception e) { + catch (@SuppressWarnings("unused") Exception e) { if (logger.isDebugEnabled()) { - logger.debug("Could not map " + k + " with type " + v.getClass().getName()); + logger.debug("Could not map " + key + " with type " + valueToAdd.getClass().getName()); } } } @@ -250,11 +251,11 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper { @Override public void toHeaders(Headers source, final Map headers) { final Map jsonTypes = decodeJsonTypes(source); - source.forEach(h -> { - if (!(h.key().equals(JSON_TYPES))) { - if (jsonTypes != null && jsonTypes.containsKey(h.key())) { + source.forEach(header -> { + if (!(header.key().equals(JSON_TYPES))) { + if (jsonTypes != null && jsonTypes.containsKey(header.key())) { Class type = Object.class; - String requestedType = jsonTypes.get(h.key()); + String requestedType = jsonTypes.get(header.key()); boolean trusted = false; try { trusted = trusted(requestedType); @@ -263,26 +264,26 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper { } } catch (Exception e) { - logger.error("Could not load class for header: " + h.key(), e); + logger.error("Could not load class for header: " + header.key(), e); } if (trusted) { try { - Object value = decodeValue(h, type); - headers.put(h.key(), value); + Object value = decodeValue(header, type); + headers.put(header.key(), value); } catch (IOException e) { - logger.error("Could not decode json type: " + new String(h.value()) + " for key: " + h + logger.error("Could not decode json type: " + new String(header.value()) + " for key: " + header .key(), e); - headers.put(h.key(), h.value()); + headers.put(header.key(), header.value()); } } else { - headers.put(h.key(), new NonTrustedHeaderType(h.value(), requestedType)); + headers.put(header.key(), new NonTrustedHeaderType(header.value(), requestedType)); } } else { - headers.put(h.key(), h.value()); + headers.put(header.key(), headertValueToAddIn(header)); } } }); @@ -419,7 +420,7 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper { return "NonTrustedHeaderType [headerValue=" + new String(this.headerValue, StandardCharsets.UTF_8) + ", untrustedType=" + this.untrustedType + "]"; } - catch (Exception e) { + catch (@SuppressWarnings("unused") Exception e) { return "NonTrustedHeaderType [headerValue=" + Arrays.toString(this.headerValue) + ", untrustedType=" + this.untrustedType + "]"; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/SimpleKafkaHeaderMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/SimpleKafkaHeaderMapper.java index 13c1caf3..5a0442fb 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/SimpleKafkaHeaderMapper.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/SimpleKafkaHeaderMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-2019 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. @@ -26,7 +26,8 @@ import org.springframework.messaging.MessageHeaders; /** * A simple header mapper that maps headers directly; for outbound, * only byte[] headers are mapped; for inbound, headers are mapped - * unchanged, as byte[]. + * unchanged, as byte[]. Strings can also be mapped to/from byte. + * See {@link #setRawMappedHaeaders(Map)}. * Most headers in {@link KafkaHeaders} are not mapped on outbound messages. * The exceptions are correlation and reply headers for request/reply * @@ -64,16 +65,17 @@ public class SimpleKafkaHeaderMapper extends AbstractKafkaHeaderMapper { @Override public void fromHeaders(MessageHeaders headers, Headers target) { - headers.forEach((k, v) -> { - if (v instanceof byte[] && matches(k, v)) { - target.add(new RecordHeader(k, (byte[]) v)); + headers.forEach((key, value) -> { + Object valueToAdd = headerValueToAddOut(key, value); + if (valueToAdd instanceof byte[] && matches(key, valueToAdd)) { + target.add(new RecordHeader(key, (byte[]) valueToAdd)); } }); } @Override public void toHeaders(Headers source, Map target) { - source.forEach(header -> target.put(header.key(), header.value())); + source.forEach(header -> target.put(header.key(), headertValueToAddIn(header))); } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java b/spring-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java index dc124284..a0ab3573 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2018 the original author or authors. + * Copyright 2017-2019 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,6 +17,7 @@ package org.springframework.kafka.support; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.entry; import java.nio.charset.Charset; import java.util.Collections; @@ -24,6 +25,8 @@ import java.util.HashMap; import java.util.Map; import java.util.Set; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeader; import org.apache.kafka.common.header.internals.RecordHeaders; import org.junit.Test; @@ -146,6 +149,85 @@ public class DefaultKafkaHeaderMapperTests { assertThat(fooHeader).isEqualTo(MimeType.valueOf("application/json")); } + @Test + public void testSpecificStringConvert() { + DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper(); + Map rawMappedHeaders = new HashMap<>(); + rawMappedHeaders.put("thisOnesAString", true); + rawMappedHeaders.put("thisOnesBytes", false); + mapper.setRawMappedHaeaders(rawMappedHeaders); + Map headersMap = new HashMap<>(); + headersMap.put("thisOnesAString", "foo"); + headersMap.put("thisOnesBytes", "bar"); + headersMap.put("alwaysRaw", "baz".getBytes()); + MessageHeaders headers = new MessageHeaders(headersMap); + Headers target = new RecordHeaders(); + mapper.fromHeaders(headers, target); + assertThat(target).containsExactlyInAnyOrder( + new RecordHeader("thisOnesAString", "foo".getBytes()), + new RecordHeader("thisOnesBytes", "bar".getBytes()), + new RecordHeader("alwaysRaw", "baz".getBytes())); + headersMap.clear(); + mapper.toHeaders(target, headersMap); + assertThat(headersMap).contains( + entry("thisOnesAString", "foo"), + entry("thisOnesBytes", "bar".getBytes()), + entry("alwaysRaw", "baz".getBytes())); + } + + @Test + public void testJsonStringConvert() { + DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper(); + Map rawMappedHeaders = new HashMap<>(); + rawMappedHeaders.put("thisOnesBytes", false); + mapper.setRawMappedHaeaders(rawMappedHeaders); + Map headersMap = new HashMap<>(); + headersMap.put("thisOnesAString", "foo"); + headersMap.put("thisOnesBytes", "bar"); + headersMap.put("alwaysRaw", "baz".getBytes()); + MessageHeaders headers = new MessageHeaders(headersMap); + Headers target = new RecordHeaders(); + mapper.fromHeaders(headers, target); + assertThat(target).containsExactlyInAnyOrder( + new RecordHeader(DefaultKafkaHeaderMapper.JSON_TYPES, + "{\"thisOnesAString\":\"java.lang.String\"}".getBytes()), + new RecordHeader("thisOnesAString", "\"foo\"".getBytes()), + new RecordHeader("alwaysRaw", "baz".getBytes()), + new RecordHeader("thisOnesBytes", "bar".getBytes())); + headersMap.clear(); + mapper.toHeaders(target, headersMap); + assertThat(headersMap).contains( + entry("thisOnesAString", "foo"), + entry("thisOnesBytes", "bar".getBytes()), + entry("alwaysRaw", "baz".getBytes())); + } + + @Test + public void testAlwaysStringConvert() { + DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper(); + mapper.setMapAllStringsOut(true); + Map rawMappedHeaders = new HashMap<>(); + rawMappedHeaders.put("thisOnesBytes", false); + mapper.setRawMappedHaeaders(rawMappedHeaders); + Map headersMap = new HashMap<>(); + headersMap.put("thisOnesAString", "foo"); + headersMap.put("thisOnesBytes", "bar"); + headersMap.put("alwaysRaw", "baz".getBytes()); + MessageHeaders headers = new MessageHeaders(headersMap); + Headers target = new RecordHeaders(); + mapper.fromHeaders(headers, target); + assertThat(target).containsExactlyInAnyOrder( + new RecordHeader("thisOnesAString", "foo".getBytes()), + new RecordHeader("thisOnesBytes", "bar".getBytes()), + new RecordHeader("alwaysRaw", "baz".getBytes())); + headersMap.clear(); + mapper.toHeaders(target, headersMap); + assertThat(headersMap).contains( + entry("thisOnesAString", "foo".getBytes()), + entry("thisOnesBytes", "bar".getBytes()), + entry("alwaysRaw", "baz".getBytes())); + } + public static final class Foo { private String bar = "bar"; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/support/SimpleKafkaHeaderMapperTests.java b/spring-kafka/src/test/java/org/springframework/kafka/support/SimpleKafkaHeaderMapperTests.java new file mode 100644 index 00000000..4d68e3e4 --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/support/SimpleKafkaHeaderMapperTests.java @@ -0,0 +1,114 @@ +/* + * Copyright 2019 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.support; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.entry; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeader; +import org.apache.kafka.common.header.internals.RecordHeaders; +import org.junit.jupiter.api.Test; + +import org.springframework.messaging.MessageHeaders; + +/** + * @author Gary Russell + * @since 2.2.5 + * + */ +public class SimpleKafkaHeaderMapperTests { + + @Test + public void testSpecificStringConvert() { + SimpleKafkaHeaderMapper mapper = new SimpleKafkaHeaderMapper("*"); + Map rawMappedHeaders = new HashMap<>(); + rawMappedHeaders.put("thisOnesAString", true); + rawMappedHeaders.put("thisOnesBytes", false); + mapper.setRawMappedHaeaders(rawMappedHeaders); + Map headersMap = new HashMap<>(); + headersMap.put("thisOnesAString", "foo"); + headersMap.put("thisOnesBytes", "bar"); + headersMap.put("neverConverted", "baz".getBytes()); + MessageHeaders headers = new MessageHeaders(headersMap); + Headers target = new RecordHeaders(); + mapper.fromHeaders(headers, target); + assertThat(target).containsExactlyInAnyOrder( + new RecordHeader("thisOnesAString", "foo".getBytes()), + new RecordHeader("thisOnesBytes", "bar".getBytes()), + new RecordHeader("neverConverted", "baz".getBytes())); + headersMap.clear(); + mapper.toHeaders(target, headersMap); + assertThat(headersMap).contains( + entry("thisOnesAString", "foo"), + entry("thisOnesBytes", "bar".getBytes()), + entry("neverConverted", "baz".getBytes())); + } + + @Test + public void testNotStringConvert() { + SimpleKafkaHeaderMapper mapper = new SimpleKafkaHeaderMapper("*"); + Map rawMappedHeaders = new HashMap<>(); + rawMappedHeaders.put("thisOnesBytes", false); + mapper.setRawMappedHaeaders(rawMappedHeaders); + Map headersMap = new HashMap<>(); + headersMap.put("thisOnesAString", "foo"); + headersMap.put("thisOnesBytes", "bar"); + headersMap.put("neverConverted", "baz".getBytes()); + MessageHeaders headers = new MessageHeaders(headersMap); + Headers target = new RecordHeaders(); + mapper.fromHeaders(headers, target); + assertThat(target).containsExactlyInAnyOrder( + new RecordHeader("neverConverted", "baz".getBytes()), + new RecordHeader("thisOnesBytes", "bar".getBytes())); + headersMap.clear(); + mapper.toHeaders(target, headersMap); + assertThat(headersMap).contains( + entry("thisOnesBytes", "bar".getBytes()), + entry("neverConverted", "baz".getBytes())); + } + + @Test + public void testAlwaysStringConvert() { + SimpleKafkaHeaderMapper mapper = new SimpleKafkaHeaderMapper("*"); + mapper.setMapAllStringsOut(true); + Map rawMappedHeaders = new HashMap<>(); + rawMappedHeaders.put("thisOnesBytes", false); + mapper.setRawMappedHaeaders(rawMappedHeaders); + Map headersMap = new HashMap<>(); + headersMap.put("thisOnesAString", "foo"); + headersMap.put("thisOnesBytes", "bar"); + headersMap.put("neverConverted", "baz".getBytes()); + MessageHeaders headers = new MessageHeaders(headersMap); + Headers target = new RecordHeaders(); + mapper.fromHeaders(headers, target); + assertThat(target).containsExactlyInAnyOrder( + new RecordHeader("thisOnesAString", "foo".getBytes()), + new RecordHeader("thisOnesBytes", "bar".getBytes()), + new RecordHeader("neverConverted", "baz".getBytes())); + headersMap.clear(); + mapper.toHeaders(target, headersMap); + assertThat(headersMap).contains( + entry("thisOnesAString", "foo".getBytes()), + entry("thisOnesBytes", "bar".getBytes()), + entry("neverConverted", "baz".getBytes())); + } + +} diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index de138c16..dbfb88aa 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -2482,6 +2482,47 @@ You can trust other (or all) packages by adding trusted packages with the `addTr If you receive messages from untrusted sources, you may wish to add only those packages you trust. To trust all packages, you can use `mapper.addTrustedPackages("*")`. +NOTE: Mapping `String` header values in a raw form is useful when communicating with systems that are not aware of the mapper's JSON format. + +Starting with version 2.2.5, you can specify that certain string-valued headers should not be mapped using JSON, but to/from a raw `byte[]`. +The `AbstractKafkaHeaderMapper` has new properties; `mapAllStringsOut` when set to true, all string-valued headers will be converted to `byte[]` using the `charset` property (default `UTF-8`). +In addition, there is a property `rawMappedHeaders`, which is a map of `header name : boolean`; if the map contains a header name, and the header contains a `String` value, it will be mapped as a raw `byte[]` using the charset. +This map is also used to map raw incoming `byte[]` headers to `String` using the charset if, and only if, the boolean in the map value is `true`. +If the boolean is `false`, or the header name is not in the map with a `true` value, the incoming header is simply mapped as the raw unmapped header. + +The following test case illustrates this mechanism. + +==== +[source, java] +---- +@Test +public void testSpecificStringConvert() { + DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper(); + Map rawMappedHeaders = new HashMap<>(); + rawMappedHeaders.put("thisOnesAString", true); + rawMappedHeaders.put("thisOnesBytes", false); + mapper.setRawMappedHaeaders(rawMappedHeaders); + Map headersMap = new HashMap<>(); + headersMap.put("thisOnesAString", "thing1"); + headersMap.put("thisOnesBytes", "thing2"); + headersMap.put("alwaysRaw", "thing3".getBytes()); + MessageHeaders headers = new MessageHeaders(headersMap); + Headers target = new RecordHeaders(); + mapper.fromHeaders(headers, target); + assertThat(target).containsExactlyInAnyOrder( + new RecordHeader("thisOnesAString", "thing1".getBytes()), + new RecordHeader("thisOnesBytes", "thing2".getBytes()), + new RecordHeader("alwaysRaw", "thing3".getBytes())); + headersMap.clear(); + mapper.toHeaders(target, headersMap); + assertThat(headersMap).contains( + entry("thisOnesAString", "thing1"), + entry("thisOnesBytes", "thing2".getBytes()), + entry("alwaysRaw", "thing3".getBytes())); +} +---- +==== + By default, the `DefaultKafkaHeaderMapper` is used in the `MessagingMessageConverter` and `BatchMessagingMessageConverter`, as long as Jackson is on the class path. With the batch converter, the converted headers are available in the `KafkaHeaders.BATCH_CONVERTED_HEADERS` as a `List>` where the map in a position of the list corresponds to the data position in the payload.