GH-769: Add JSON Type Mapping Property

Resolves #769

Allow configuration of type mapping using Kafka properties.

GH-768: Override JSON Type Headers

Resolves #768

Polishing - PR Comments

* Fix Copyrights
* Some code improvements in `Json(De)Serializer`

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/support/converter/DefaultJackson2JavaTypeMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -50,7 +50,7 @@ public class DefaultJackson2JavaTypeMapper extends AbstractJavaTypeMapper
 					"java.lang"
 			);
 
-	private final Set<String> trustedPackages = new LinkedHashSet<String>(TRUSTED_PACKAGES);
+	private final Set<String> trustedPackages = new LinkedHashSet<>(TRUSTED_PACKAGES);
 
 	private volatile TypePrecedence typePrecedence = TypePrecedence.INFERRED;
 
@@ -64,21 +64,7 @@ public TypePrecedence getTypePrecedence() {
 		return this.typePrecedence;
 	}
 
-	/**
-	 * Set the precedence for evaluating type information in message properties.
-	 * When using {@code @KafkaListener} at the method level, the framework attempts
-	 * to determine the target type for payload conversion from the method signature.
-	 * If so, this type is provided by the {@code MessagingMessageListenerAdapter}.
-	 * <p>
-	 * By default, if the type is concrete (not abstract, not an interface), this will
-	 * be used ahead of type information provided in the {@code __TypeId__} and
-	 * associated headers provided by the sender.
-	 * <p>
-	 * If you wish to force the use of the  {@code __TypeId__} and associated headers
-	 * (such as when the actual type is a subclass of the method argument type),
-	 * set the precedence to {@link Jackson2JavaTypeMapper.TypePrecedence#TYPE_ID}.
-	 * @param typePrecedence the precedence.
-	 */
+	@Override
 	public void setTypePrecedence(TypePrecedence typePrecedence) {
 		Assert.notNull(typePrecedence, "'typePrecedence' cannot be null");
 		this.typePrecedence = typePrecedence;

spring-kafka/src/main/java/org/springframework/kafka/support/converter/Jackson2JavaTypeMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -48,6 +48,24 @@ enum TypePrecedence {
 
 	TypePrecedence getTypePrecedence();
 
+	/**
+	 * Set the precedence for evaluating type information in message properties.
+	 * When using {@code @KafkaListener} at the method level, the framework attempts
+	 * to determine the target type for payload conversion from the method signature.
+	 * If so, this type is provided by the {@code MessagingMessageListenerAdapter}.
+	 * <p> By default, if the type is concrete (not abstract, not an interface), this will
+	 * be used ahead of type information provided in the {@code __TypeId__} and
+	 * associated headers provided by the sender.
+	 * <p> If you wish to force the use of the  {@code __TypeId__} and associated headers
+	 * (such as when the actual type is a subclass of the method argument type),
+	 * set the precedence to {@link Jackson2JavaTypeMapper.TypePrecedence#TYPE_ID}.
+	 * @param typePrecedence the precedence.
+	 * @since 2.2
+	 */
+	default void setTypePrecedence(TypePrecedence typePrecedence) {
+		throw new UnsupportedOperationException("This mapper does not support this method");
+	}
+
 	void addTrustedPackages(String... packages);
 
 }

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java

@@ -29,6 +29,8 @@
 import org.springframework.kafka.support.converter.AbstractJavaTypeMapper;
 import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper;
 import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper;
+import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper.TypePrecedence;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
@@ -81,6 +83,12 @@ public class JsonDeserializer<T> implements ExtendedDeserializer<T> {
 	 */
 	public static final String TRUSTED_PACKAGES = "spring.json.trusted.packages";
 
+	/**
+	 * Kafka config property to add type mappings to the type mapper:
+	 * 'foo=com.Foo,bar=com.Bar'.
+	 */
+	public static final String TYPE_MAPPINGS = JsonSerializer.TYPE_MAPPINGS;
+
 	protected final ObjectMapper objectMapper;
 
 	protected Class<T> targetType;
@@ -91,29 +99,79 @@ public class JsonDeserializer<T> implements ExtendedDeserializer<T> {
 
 	private boolean typeMapperExplicitlySet = false;
 
+	/**
+	 * Construct an instance with a default {@link ObjectMapper}.
+	 */
 	public JsonDeserializer() {
-		this((Class<T>) null);
+		this(null, true);
 	}
 
-	protected JsonDeserializer(ObjectMapper objectMapper) {
-		this(null, objectMapper);
+	/**
+	 * Construct an instance with the provided {@link ObjectMapper}.
+	 * @param objectMapper a custom object mapper.
+	 */
+	public JsonDeserializer(ObjectMapper objectMapper) {
+		this(null, objectMapper, true);
 	}
 
+	/**
+	 * Construct an instance with the provided target type, and a default
+	 * {@link ObjectMapper}.
+	 * @param targetType the target type to use if no type info headers are present.
+	 */
 	public JsonDeserializer(Class<T> targetType) {
-		this(targetType, new ObjectMapper());
+		this(targetType, true);
+	}
+
+	/**
+	 * Construct an instance with the provided target type, and
+	 * useHeadersIfPresent with a default {@link ObjectMapper}.
+	 * @param targetType the target type.
+	 * @param useHeadersIfPresent true to use headers if present and fall back to target
+	 * type if not.
+	 * @since 2.2
+	 */
+	public JsonDeserializer(Class<T> targetType, boolean useHeadersIfPresent) {
+		this(targetType, new ObjectMapper(), useHeadersIfPresent);
 		this.objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
 		this.objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
 	}
 
-	@SuppressWarnings("unchecked")
+	/**
+	 * Construct an instance with the provided target type, and {@link ObjectMapper}.
+	 * @param targetType the target type to use if no type info headers are present.
+	 * @param objectMapper the mapper. type if not.
+	 */
 	public JsonDeserializer(Class<T> targetType, ObjectMapper objectMapper) {
+		this(targetType, objectMapper, true);
+	}
+
+	/**
+	 * Construct an instance with the provided target type, {@link ObjectMapper} and
+	 * useHeadersIfPresent.
+	 * @param targetType the target type.
+	 * @param objectMapper the mapper.
+	 * @param useHeadersIfPresent true to use headers if present and fall back to target
+	 * type if not.
+	 * @since 2.2
+	 */
+	@SuppressWarnings("unchecked")
+	public JsonDeserializer(@Nullable Class<T> targetType, ObjectMapper objectMapper, boolean useHeadersIfPresent) {
 		Assert.notNull(objectMapper, "'objectMapper' must not be null.");
 		this.objectMapper = objectMapper;
-		if (targetType == null) {
-			targetType = (Class<T>) ResolvableType.forClass(getClass()).getSuperType().resolveGeneric(0);
-		}
 		this.targetType = targetType;
+		if (this.targetType == null) {
+			this.targetType = (Class<T>) ResolvableType.forClass(getClass()).getSuperType().resolveGeneric(0);
+		}
+		Assert.isTrue(this.targetType != null || useHeadersIfPresent,
+				"'targetType' cannot be null if 'useHeadersIfPresent' is false");
+
+		if (this.targetType != null) {
+			this.reader = this.objectMapper.readerFor(this.targetType);
+		}
+
 		addTargetPackageToTrusted();
+		this.typeMapper.setTypePrecedence(useHeadersIfPresent ? TypePrecedence.TYPE_ID : TypePrecedence.INFERRED);
 	}
 
 	public Jackson2JavaTypeMapper getTypeMapper() {
@@ -185,6 +243,10 @@ else if (configs.get(VALUE_DEFAULT_TYPE) instanceof String) {
 					throw new IllegalStateException(VALUE_DEFAULT_TYPE + " must be Class or String");
 				}
 			}
+
+			if (this.targetType != null) {
+				this.reader = this.objectMapper.readerFor(this.targetType);
+			}
 			addTargetPackageToTrusted();
 		}
 		catch (ClassNotFoundException | LinkageError e) {
@@ -196,6 +258,11 @@ else if (configs.get(VALUE_DEFAULT_TYPE) instanceof String) {
 						StringUtils.commaDelimitedListToStringArray((String) configs.get(TRUSTED_PACKAGES)));
 			}
 		}
+		if (configs.containsKey(TYPE_MAPPINGS) && !this.typeMapperExplicitlySet
+				&& this.typeMapper instanceof AbstractJavaTypeMapper) {
+			((AbstractJavaTypeMapper) this.typeMapper).setIdClassMapping(
+					JsonSerializer.createMappings((String) configs.get(JsonSerializer.TYPE_MAPPINGS)));
+		}
 	}
 
 	/**
@@ -218,30 +285,32 @@ public T deserialize(String topic, Headers headers, byte[] data) {
 		if (data == null) {
 			return null;
 		}
-		JavaType javaType = this.typeMapper.toJavaType(headers);
-		if (javaType == null) {
-			Assert.state(this.targetType != null, "No type information in headers and no default type provided");
-			return deserialize(topic, data);
-		}
-		else {
-			try {
-				return this.objectMapper.readerFor(javaType).readValue(data);
-			}
-			catch (IOException e) {
-				throw new SerializationException("Can't deserialize data [" + Arrays.toString(data) +
-						"] from topic [" + topic + "]", e);
+		ObjectReader reader = null;
+		if (this.typeMapper.getTypePrecedence().equals(TypePrecedence.TYPE_ID)) {
+			JavaType javaType = this.typeMapper.toJavaType(headers);
+			if (javaType != null) {
+				reader = this.objectMapper.readerFor(javaType);
 			}
 		}
+		if (reader == null) {
+			reader = this.reader;
+		}
+		Assert.state(reader != null, "No type information in headers and no default type provided");
+		try {
+			return reader.readValue(data);
+		}
+		catch (IOException e) {
+			throw new SerializationException("Can't deserialize data [" + Arrays.toString(data) +
+					"] from topic [" + topic + "]", e);
+		}
 	}
 
 	@Override
 	public T deserialize(String topic, byte[] data) {
 		if (data == null) {
 			return null;
 		}
-		if (this.reader == null) {
-			this.reader = this.objectMapper.readerFor(this.targetType);
-		}
+		Assert.state(this.reader != null, "No headers available and no default type provided");
 		try {
 			T result = null;
 			if (data != null) {
@@ -259,4 +328,5 @@ public T deserialize(String topic, byte[] data) {
 	public void close() {
 		// No-op
 	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java

@@ -17,6 +17,7 @@
 package org.springframework.kafka.support.serializer;
 
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.Map;
 
 import org.apache.kafka.common.errors.SerializationException;
@@ -28,6 +29,8 @@
 import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper;
 import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.StringUtils;
 
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.MapperFeature;
@@ -46,10 +49,16 @@
 public class JsonSerializer<T> implements ExtendedSerializer<T> {
 
 	/**
-	 * Kafka config property for to disable adding type headers.
+	 * Kafka config property for disabling adding type headers.
 	 */
 	public static final String ADD_TYPE_INFO_HEADERS = "spring.json.add.type.headers";
 
+	/**
+	 * Kafka config property to add type mappings to the type mapper:
+	 * 'foo:com.Foo,bar:com.Bar'.
+	 */
+	public static final String TYPE_MAPPINGS = "spring.json.type.mapping";
+
 	protected final ObjectMapper objectMapper;
 
 	protected boolean addTypeInfo = true;
@@ -126,6 +135,28 @@ else if (config instanceof String) {
 				throw new IllegalStateException(ADD_TYPE_INFO_HEADERS + " must be Boolean or String");
 			}
 		}
+		if (configs.containsKey(TYPE_MAPPINGS) && !this.typeMapperExplicitlySet
+				&& this.typeMapper instanceof AbstractJavaTypeMapper) {
+			((AbstractJavaTypeMapper) this.typeMapper)
+					.setIdClassMapping(createMappings((String) configs.get(TYPE_MAPPINGS)));
+		}
+	}
+
+	protected static Map<String, Class<?>> createMappings(String mappings) {
+		Map<String, Class<?>> mappingsMap = new HashMap<>();
+		String[] array = StringUtils.commaDelimitedListToStringArray(mappings);
+		for (String entry : array) {
+			String[] split = entry.split(":");
+			Assert.isTrue(split.length == 2, "Each comma-delimited mapping entry must have exactly one ':'");
+			try {
+				mappingsMap.put(split[0].trim(),
+						ClassUtils.forName(split[1].trim(), JsonSerializer.class.getClassLoader()));
+			}
+			catch (ClassNotFoundException | LinkageError e) {
+				throw new IllegalArgumentException(e);
+			}
+		}
+		return mappingsMap;
 	}
 
 	@Override
@@ -160,4 +191,5 @@ public byte[] serialize(String topic, T data) {
 	public void close() {
 		// No-op
 	}
+
 }