GH-1012 Allow raw Strings in default header mapper

Resolves https://github.com/spring-projects/spring-kafka/issues/1012

Add configuration to map string-valued headers as raw `byte[]`
instead of adding to the map of json-mapped headers.

* Polishing - PR Comments.

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<SimplePatternBasedHeaderMatcher> matchers = new ArrayList<>(NEVER_MAPPED); // NOSONAR
 
+	private final Map<String, Boolean> 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<String, Boolean> 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.

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<String, String> 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<String, Object> headers) {
 		final Map<String, String> 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 + "]";
 			}

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<String, Object> target) {
-		source.forEach(header -> target.put(header.key(), header.value()));
+		source.forEach(header -> target.put(header.key(), headertValueToAddIn(header)));
 	}
 
 }

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<String, Boolean> rawMappedHeaders = new HashMap<>();
+		rawMappedHeaders.put("thisOnesAString", true);
+		rawMappedHeaders.put("thisOnesBytes", false);
+		mapper.setRawMappedHaeaders(rawMappedHeaders);
+		Map<String, Object> 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<String, Boolean> rawMappedHeaders = new HashMap<>();
+		rawMappedHeaders.put("thisOnesBytes", false);
+		mapper.setRawMappedHaeaders(rawMappedHeaders);
+		Map<String, Object> 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<String, Boolean> rawMappedHeaders = new HashMap<>();
+		rawMappedHeaders.put("thisOnesBytes", false);
+		mapper.setRawMappedHaeaders(rawMappedHeaders);
+		Map<String, Object> 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";

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<String, Boolean> rawMappedHeaders = new HashMap<>();
+		rawMappedHeaders.put("thisOnesAString", true);
+		rawMappedHeaders.put("thisOnesBytes", false);
+		mapper.setRawMappedHaeaders(rawMappedHeaders);
+		Map<String, Object> 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<String, Boolean> rawMappedHeaders = new HashMap<>();
+		rawMappedHeaders.put("thisOnesBytes", false);
+		mapper.setRawMappedHaeaders(rawMappedHeaders);
+		Map<String, Object> 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<String, Boolean> rawMappedHeaders = new HashMap<>();
+		rawMappedHeaders.put("thisOnesBytes", false);
+		mapper.setRawMappedHaeaders(rawMappedHeaders);
+		Map<String, Object> 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()));
+	}
+
+}

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<String, Boolean> rawMappedHeaders = new HashMap<>();
+    rawMappedHeaders.put("thisOnesAString", true);
+    rawMappedHeaders.put("thisOnesBytes", false);
+    mapper.setRawMappedHaeaders(rawMappedHeaders);
+    Map<String, Object> 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<Map<String, Object>>` where the map in a position of the list corresponds to the data position in the payload.