GH-1012 Allow raw Strings in default header mapper

Resolves #1012

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

* Polishing - PR Comments.

Author: garyrussell

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,13 +65,62 @@ 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) {
 			this.matchers.add(new SimplePatternBasedHeaderMatcher(pattern));
 		}
 	}
 
+	/**
+	 * 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 @@ protected boolean matches(String header) {
 		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 @@ protected Set<String> getToStringClasses() {
 	 * 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 void addToStringClasses(String... classNames) {
 	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 void fromHeaders(MessageHeaders headers, Headers target) {
 	@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 void toHeaders(Headers source, final Map<String, Object> headers) {
 						}
 					}
 					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 String toString() {
 				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 @@
 /**
  * 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 SimpleKafkaHeaderMapper(String... patterns) {
 
 	@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)));
 	}
 
 }