DefaultKafkaHeaderMapper Improvement

String-valued headers are now mapped as raw strings by default instead
of being encoded as JSON strings.

This can be disabled until all applications have been upgraded to use
2.3 or later.

* Fix use of `encodeStrings`; remove bogus `SuppressWarnings`.

Author: garyrussell

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

@@ -20,7 +20,6 @@
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.HashMap;
-import java.util.Iterator;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
@@ -60,6 +59,8 @@
  */
 public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper {
 
+	private static final String JAVA_LANG_STRING = "java.lang.String";
+
 	private static final List<String> DEFAULT_TRUSTED_PACKAGES =
 			Arrays.asList(
 					"java.lang",
@@ -79,6 +80,8 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper {
 
 	private final Set<String> toStringClasses = new LinkedHashSet<>();
 
+	private boolean encodeStrings;
+
 	/**
 	 * Construct an instance with the default object mapper and default header patterns
 	 * for outbound headers; all inbound headers are mapped. The default pattern list is
@@ -171,6 +174,22 @@ protected Set<String> getToStringClasses() {
 		return this.toStringClasses;
 	}
 
+	protected boolean isEncodeStrings() {
+		return this.encodeStrings;
+	}
+
+	/**
+	 * Set to true to encode String-valued headers as JSON ("..."), by default just the
+	 * raw String value is converted to a byte array using the configured charset. Set to
+	 * true if a consumer of the outbound record is using Spring for Apache Kafka version
+	 * less than 2.3
+	 * @param encodeStrings true to encode (default false).
+	 * @since 2.3
+	 */
+	public void setEncodeStrings(boolean encodeStrings) {
+		this.encodeStrings = encodeStrings;
+	}
+
 	/**
 	 * Add packages to the trusted packages list (default {@code java.util, java.lang}) used
 	 * when constructing objects from JSON.
@@ -207,9 +226,9 @@ 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((key, val) -> {
-			if (matches(key, val)) {
-				Object valueToAdd = headerValueToAddOut(key, val);
+		headers.forEach((key, rawValue) -> {
+			if (matches(key, rawValue)) {
+				Object valueToAdd = headerValueToAddOut(key, rawValue);
 				if (valueToAdd instanceof byte[]) {
 					target.add(new RecordHeader(key, (byte[]) valueToAdd));
 				}
@@ -218,14 +237,20 @@ public void fromHeaders(MessageHeaders headers, Headers target) {
 						Object value = valueToAdd;
 						String className = valueToAdd.getClass().getName();
 						if (this.toStringClasses.contains(className)) {
-							value = valueToAdd.toString();
-							className = "java.lang.String";
+							valueToAdd = valueToAdd.toString();
+							className = JAVA_LANG_STRING;
+						}
+						if (!this.encodeStrings && valueToAdd instanceof String) {
+							target.add(new RecordHeader(key, ((String) valueToAdd).getBytes(getCharset())));
+							className = JAVA_LANG_STRING;
+						}
+						else {
+							target.add(new RecordHeader(key, headerObjectMapper.writeValueAsBytes(value)));
 						}
-						target.add(new RecordHeader(key, headerObjectMapper.writeValueAsBytes(value)));
 						jsonHeaders.put(key, className);
 					}
 					catch (Exception e) {
-						logger.debug(e, () -> "Could not map " + key + " with type " + valueToAdd.getClass().getName());
+						logger.debug(e, () -> "Could not map " + key + " with type " + rawValue.getClass().getName());
 					}
 				}
 			}
@@ -258,20 +283,25 @@ public void toHeaders(Headers source, final Map<String, Object> headers) {
 					catch (Exception e) {
 						logger.error(e, () -> "Could not load class for header: " + header.key());
 					}
-					if (trusted) {
-						try {
-							Object value = decodeValue(header, type);
-							headers.put(header.key(), value);
-						}
-						catch (IOException e) {
-							logger.error(e, () ->
-									"Could not decode json type: " + new String(header.value()) + " for key: "
-											+ header.key());
-							headers.put(header.key(), header.value());
-						}
+					if (String.class.equals(type) && header.value().length > 0 && header.value()[0] != '"') {
+						headers.put(header.key(), new String(header.value(), getCharset()));
 					}
 					else {
-						headers.put(header.key(), new NonTrustedHeaderType(header.value(), requestedType));
+						if (trusted) {
+							try {
+								Object value = decodeValue(header, type);
+								headers.put(header.key(), value);
+							}
+							catch (IOException e) {
+								logger.error(e, () ->
+										"Could not decode json type: " + new String(header.value()) + " for key: "
+												+ header.key());
+								headers.put(header.key(), header.value());
+							}
+						}
+						else {
+							headers.put(header.key(), new NonTrustedHeaderType(header.value(), requestedType));
+						}
 					}
 				}
 				else {
@@ -304,18 +334,14 @@ private Object decodeValue(Header h, Class<?> type) throws IOException, LinkageE
 	@Nullable
 	private Map<String, String> decodeJsonTypes(Headers source) {
 		Map<String, String> types = null;
-		Iterator<Header> iterator = source.iterator();
-		ObjectMapper headerObjectMapper = getObjectMapper();
-		while (iterator.hasNext()) {
-			Header next = iterator.next();
-			if (next.key().equals(JSON_TYPES)) {
-				try {
-					types = headerObjectMapper.readValue(next.value(), Map.class);
-				}
-				catch (IOException e) {
-					logger.error(e, () -> "Could not decode json types: " + new String(next.value()));
-				}
-				break;
+		Header jsonTypes = source.lastHeader(JSON_TYPES);
+		if (jsonTypes != null) {
+			ObjectMapper headerObjectMapper = getObjectMapper();
+			try {
+				types = headerObjectMapper.readValue(jsonTypes.value(), Map.class);
+			}
+			catch (IOException e) {
+				logger.error(e, () -> "Could not decode json types: " + new String(jsonTypes.value()));
 			}
 		}
 		return types;

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java

@@ -208,7 +208,7 @@ public void testWithMessage() {
 		assertThat(iterator.hasNext()).isTrue();
 		Header next = iterator.next();
 		assertThat(next.key()).isEqualTo("foo");
-		assertThat(new String(next.value())).isEqualTo("\"bar\"");
+		assertThat(new String(next.value())).isEqualTo("bar");
 		assertThat(iterator.hasNext()).isTrue();
 		next = iterator.next();
 		assertThat(next.key()).isEqualTo(DefaultKafkaHeaderMapper.JSON_TYPES);

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

@@ -188,15 +188,29 @@ public void testJsonStringConvert() {
 		assertThat(target).containsExactlyInAnyOrder(
 				new RecordHeader(DefaultKafkaHeaderMapper.JSON_TYPES,
 						"{\"thisOnesAString\":\"java.lang.String\"}".getBytes()),
-				new RecordHeader("thisOnesAString", "\"foo\"".getBytes()),
+				new RecordHeader("thisOnesAString", "foo".getBytes()),
 				new RecordHeader("alwaysRaw", "baz".getBytes()),
 				new RecordHeader("thisOnesBytes", "bar".getBytes()));
 		headersMap.clear();
+		target.add(new RecordHeader(DefaultKafkaHeaderMapper.JSON_TYPES,
+						("{\"thisOnesAString\":\"java.lang.String\","
+						+ "\"backwardCompatible\":\"java.lang.String\"}").getBytes()));
+		target.add(new RecordHeader("backwardCompatible", "\"qux\"".getBytes()));
 		mapper.toHeaders(target, headersMap);
 		assertThat(headersMap).contains(
 				entry("thisOnesAString", "foo"),
 				entry("thisOnesBytes", "bar".getBytes()),
-				entry("alwaysRaw", "baz".getBytes()));
+				entry("alwaysRaw", "baz".getBytes()),
+				entry("backwardCompatible", "qux"));
+		mapper.setEncodeStrings(true);
+		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()));
 	}
 
 	@Test

spring-kafka/src/test/java/org/springframework/kafka/support/serializer/DelegatingSerializationTests.java

@@ -98,7 +98,7 @@ private void doTest(DelegatingSerializer serializer, DelegatingDeserializer dese
 				Collections.singletonMap(DelegatingSerializer.SERIALIZATION_SELECTOR, "string"));
 		new DefaultKafkaHeaderMapper().fromHeaders(messageHeaders, headers);
 		assertThat(headers.lastHeader(DelegatingSerializer.SERIALIZATION_SELECTOR).value())
-				.isEqualTo(new byte[] { '"', 's', 't', 'r', 'i', 'n', 'g', '"' });
+				.isEqualTo(new byte[] { 's', 't', 'r', 'i', 'n', 'g' });
 		serialized = serializer.serialize("foo", headers, "bar");
 		assertThat(serialized).isEqualTo(new byte[] { 'b', 'a', 'r' });
 		assertThat(deserializer.deserialize("foo", headers, serialized)).isEqualTo("bar");

src/reference/asciidoc/kafka.adoc

@@ -3221,6 +3221,14 @@ The `DefaultKafkaHeaderMapper` has a method called `addToStringClasses()` that l
 During inbound mapping, they are mapped as `String`.
 By default, only `org.springframework.util.MimeType` and `org.springframework.http.MediaType` are mapped this way.
 
+NOTE: Starting with version 2.3, handling of String-valued headers is simplified.
+Such headers are no longer JSON encoded, by default (i.e. they do not have enclosing `"..."` added).
+The type is still added to the JSON_TYPES header so the receiving system can convert back to a String (from `byte[]`).
+The mapper can handle (decode) headers produced by older versions (it checks for a leading `"`); in this way an application using 2.3 can consume records from older versions.
+
+IMPORTANT: To be compatible with earlier versions, set `encodeStrings` to `true`, if records produced by a version using 2.3 might be consumed by applications using earlier versions.
+When all applications are using 2.3 or higher, you can leave the property at its default value of `false`.
+
 [[tombstones]]
 ==== Null Payloads and Log Compaction of 'Tombstone' Records
 

src/reference/asciidoc/whats-new.adoc

@@ -12,7 +12,7 @@ Please submit GitHub issues and/or pull requests for additional entries in that
 [[kafka-client-2.2]]
 ==== Kafka Client Version
 
-This version requires the 2.2.0 `kafka-clients` or higher.
+This version requires the 2.3.0 `kafka-clients` or higher.
 
 ==== Class/Package Changes
 
@@ -148,3 +148,8 @@ See <<kafka-testing-embeddedkafka-annotation>> for more information.
 
 You can now customize the header names for correlation, reply topic and reply partition.
 See <<replying-template>> for more information.
+
+==== Header Mapper Changes
+
+The `DefaultKafkaHeaderMapper` no longer encodes simple String-valued headers as JSON.
+See <<header-mapping>> for more information.