GH-492: JSON Improvements

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

For `StringJsonMessageConverter` and `JsonSerializer`

- Convey type information in Headers using a Jackson type mapper
- Setting `addTypeInfo` to `false` disables this feature

For `StringJsonMessageConverter` (inbound)

- If configured with with the default type mapper, no changes - use the inferred type from the method signature
- If configured with a type mapper where the `TypePrecedence` is not `INFERRED`, use the headers if available.

For `JsonDeserializer`

- use type information from headers, falling back to a default if present

For `JsonSerializer` and `JsonDeserializer`

- provide configuration (e.g. default type) via kafka properties

Polishing - @since

* Fix `DefaultJackson2JavaTypeMapper` JavaDoc do not mention Spring AMQP

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

@@ -0,0 +1,136 @@
+/*
+ * Copyright 2017 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.converter;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeader;
+
+import org.springframework.beans.factory.BeanClassLoaderAware;
+import org.springframework.messaging.converter.MessageConversionException;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Abstract type mapper.
+ *
+ * @author Mark Pollack
+ * @author Sam Nelson
+ * @author Andreas Asplund
+ * @author Gary Russell
+ *
+ * @since 2.1
+ */
+public abstract class AbstractJavaTypeMapper implements BeanClassLoaderAware {
+
+	/**
+	 * Default header name for type information.
+	 */
+	public static final String DEFAULT_CLASSID_FIELD_NAME = "__TypeId__";
+
+	/**
+	 * Default header name for container object contents type information.
+	 */
+	public static final String DEFAULT_CONTENT_CLASSID_FIELD_NAME = "__ContentTypeId__";
+
+	/**
+	 * Default header name for map key type information.
+	 */
+	public static final String DEFAULT_KEY_CLASSID_FIELD_NAME = "__KeyTypeId__";
+
+	private final Map<String, Class<?>> idClassMapping = new HashMap<String, Class<?>>();
+
+	private final Map<Class<?>, byte[]> classIdMapping = new HashMap<Class<?>, byte[]>();
+
+	private ClassLoader classLoader = ClassUtils.getDefaultClassLoader();
+
+	public String getClassIdFieldName() {
+		return DEFAULT_CLASSID_FIELD_NAME;
+	}
+
+	public String getContentClassIdFieldName() {
+		return DEFAULT_CONTENT_CLASSID_FIELD_NAME;
+	}
+
+	public String getKeyClassIdFieldName() {
+		return DEFAULT_KEY_CLASSID_FIELD_NAME;
+	}
+
+	public void setIdClassMapping(Map<String, Class<?>> idClassMapping) {
+		this.idClassMapping.putAll(idClassMapping);
+		createReverseMap();
+	}
+
+	@Override
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		this.classLoader = classLoader;
+	}
+
+	protected ClassLoader getClassLoader() {
+		return this.classLoader;
+	}
+
+	protected void addHeader(Headers headers, String headerName, Class<?> clazz) {
+		if (this.classIdMapping.containsKey(clazz)) {
+			headers.add(new RecordHeader(headerName, this.classIdMapping.get(clazz)));
+		}
+		else {
+			headers.add(new RecordHeader(headerName, clazz.getName().getBytes(StandardCharsets.UTF_8)));
+		}
+	}
+
+	protected String retrieveHeader(Headers headers, String headerName) {
+		String classId = retrieveHeaderAsString(headers, headerName);
+		if (classId == null) {
+			throw new MessageConversionException(
+					"failed to convert Message content. Could not resolve " + headerName + " in header");
+		}
+		return classId;
+	}
+
+	protected String retrieveHeaderAsString(Headers headers, String headerName) {
+		Iterator<Header> headerValues = headers.headers(headerName).iterator();
+		if (headerValues.hasNext()) {
+			Header headerValue = headerValues.next();
+			String classId = null;
+			if (headerValue.value() != null) {
+				classId = new String(headerValue.value(), StandardCharsets.UTF_8);
+			}
+			return classId;
+		}
+		return null;
+	}
+
+	private void createReverseMap() {
+		this.classIdMapping.clear();
+		for (Map.Entry<String, Class<?>> entry : this.idClassMapping.entrySet()) {
+			String id = entry.getKey();
+			Class<?> clazz = entry.getValue();
+			this.classIdMapping.put(clazz, id.getBytes(StandardCharsets.UTF_8));
+		}
+	}
+
+	public Map<String, Class<?>> getIdClassMapping() {
+		return Collections.unmodifiableMap(this.idClassMapping);
+	}
+
+}

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2017 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.converter;
+
+import org.apache.kafka.common.header.Headers;
+
+/**
+ * Strategy for setting metadata on messages such that one can create the class
+ * that needs to be instantiated when receiving a message.
+ *
+ * @author Mark Pollack
+ * @author James Carr
+ * @author Gary Russell
+ *
+ * @since 2.1
+ */
+public interface ClassMapper {
+
+	void fromClass(Class<?> clazz, Headers headers);
+
+	Class<?> toClass(Headers headers);
+
+}

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

@@ -0,0 +1,199 @@
+/*
+ * Copyright 2017 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.converter;
+
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.kafka.common.header.Headers;
+
+import org.springframework.messaging.converter.MessageConversionException;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.type.TypeFactory;
+
+/**
+ * Jackson 2 type mapper.
+ *
+ * @author Mark Pollack
+ * @author Sam Nelson
+ * @author Andreas Asplund
+ * @author Artem Bilan
+ * @author Gary Russell
+ *
+ * @since 2.1
+ */
+public class DefaultJackson2JavaTypeMapper extends AbstractJavaTypeMapper
+		implements Jackson2JavaTypeMapper, ClassMapper {
+
+	private static final List<String> TRUSTED_PACKAGES =
+			Arrays.asList(
+					"java.util",
+					"java.lang"
+			);
+
+	private final Set<String> trustedPackages = new LinkedHashSet<String>(TRUSTED_PACKAGES);
+
+	private volatile TypePrecedence typePrecedence = TypePrecedence.INFERRED;
+
+	/**
+	 * Return the precedence.
+	 * @return the precedence.
+	 * @see #setTypePrecedence(Jackson2JavaTypeMapper.TypePrecedence)
+	 */
+	@Override
+	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.
+	 */
+	public void setTypePrecedence(TypePrecedence typePrecedence) {
+		Assert.notNull(typePrecedence, "'typePrecedence' cannot be null");
+		this.typePrecedence = typePrecedence;
+	}
+
+	/**
+	 * Specify a set of packages to trust during deserialization.
+	 * The asterisk ({@code *}) means trust all.
+	 * @param trustedPackages the trusted Java packages for deserialization
+	 */
+	@Override
+	public void addTrustedPackages(String... trustedPackages) {
+		if (trustedPackages != null) {
+			for (String whiteListClass : trustedPackages) {
+				if ("*".equals(whiteListClass)) {
+					this.trustedPackages.clear();
+					break;
+				}
+				else {
+					this.trustedPackages.add(whiteListClass);
+				}
+			}
+		}
+	}
+
+	@Override
+	public JavaType toJavaType(Headers headers) {
+		String typeIdHeader = retrieveHeaderAsString(headers, getClassIdFieldName());
+
+		if (typeIdHeader != null) {
+
+			JavaType classType = getClassIdType(typeIdHeader);
+			if (!classType.isContainerType() || classType.isArrayType()) {
+				return classType;
+			}
+
+			JavaType contentClassType = getClassIdType(retrieveHeader(headers, getContentClassIdFieldName()));
+			if (classType.getKeyType() == null) {
+				return TypeFactory.defaultInstance()
+						.constructCollectionLikeType(classType.getRawClass(), contentClassType);
+			}
+
+			JavaType keyClassType = getClassIdType(retrieveHeader(headers, getKeyClassIdFieldName()));
+			return TypeFactory.defaultInstance()
+					.constructMapLikeType(classType.getRawClass(), keyClassType, contentClassType);
+		}
+
+		return null;
+	}
+
+	private JavaType getClassIdType(String classId) {
+		if (getIdClassMapping().containsKey(classId)) {
+			return TypeFactory.defaultInstance().constructType(getIdClassMapping().get(classId));
+		}
+		else {
+			try {
+				if (!isTrustedPackage(classId)) {
+					throw new IllegalArgumentException("The class '" + classId
+							+ "' is not in the trusted packages: "
+							+ this.trustedPackages + ". "
+							+ "If you believe this class is safe to deserialize, please provide its name. "
+							+ "If the serialization is only done by a trusted source, you can also enable "
+							+ "trust all (*).");
+				}
+				else {
+					return TypeFactory.defaultInstance()
+							.constructType(ClassUtils.forName(classId, getClassLoader()));
+				}
+			}
+			catch (ClassNotFoundException e) {
+				throw new MessageConversionException("failed to resolve class name. Class not found ["
+						+ classId + "]", e);
+			}
+			catch (LinkageError e) {
+				throw new MessageConversionException("failed to resolve class name. Linkage error ["
+						+ classId + "]", e);
+			}
+		}
+	}
+
+	private boolean isTrustedPackage(String requestedType) {
+		if (!this.trustedPackages.isEmpty()) {
+			String packageName = ClassUtils.getPackageName(requestedType).replaceFirst("\\[L", "");
+			for (String trustedPackage : this.trustedPackages) {
+				if (packageName.equals(trustedPackage)) {
+					return true;
+				}
+			}
+			return false;
+		}
+		return true;
+	}
+
+	@Override
+	public void fromJavaType(JavaType javaType, Headers headers) {
+		addHeader(headers, getClassIdFieldName(), javaType.getRawClass());
+
+		if (javaType.isContainerType() && !javaType.isArrayType()) {
+			addHeader(headers, getContentClassIdFieldName(), javaType.getContentType().getRawClass());
+		}
+
+		if (javaType.getKeyType() != null) {
+			addHeader(headers, getKeyClassIdFieldName(), javaType.getKeyType().getRawClass());
+		}
+	}
+
+	@Override
+	public void fromClass(Class<?> clazz, Headers headers) {
+		fromJavaType(TypeFactory.defaultInstance().constructType(clazz), headers);
+
+	}
+
+	@Override
+	public Class<?> toClass(Headers headers) {
+		return toJavaType(headers).getRawClass();
+	}
+
+}

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

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2017 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.converter;
+
+import org.apache.kafka.common.header.Headers;
+
+import com.fasterxml.jackson.databind.JavaType;
+
+/**
+ * Strategy for setting metadata on messages such that one can create the class that needs
+ * to be instantiated when receiving a message.
+ *
+ * @author Mark Pollack
+ * @author James Carr
+ * @author Sam Nelson
+ * @author Andreas Asplund
+ * @author Gary Russell
+ *
+ * @since 2.1
+ */
+public interface Jackson2JavaTypeMapper extends ClassMapper {
+
+	/**
+	 * The precedence for type conversion - inferred from the method parameter or message
+	 * headers. Only applies if both exist.
+	 */
+	enum TypePrecedence {
+		INFERRED, TYPE_ID
+	}
+
+	void fromJavaType(JavaType javaType, Headers headers);
+
+	JavaType toJavaType(Headers headers);
+
+	TypePrecedence getTypePrecedence();
+
+	void addTrustedPackages(String... packages);
+
+}

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

@@ -24,6 +24,7 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
 
 import org.springframework.kafka.support.Acknowledgment;
@@ -136,12 +137,22 @@ public class MessagingMessageConverter implements RecordMessageConverter {
 		Object key = headers.get(KafkaHeaders.MESSAGE_KEY);
 		Object payload = convertPayload(message);
 		Long timestamp = headers.get(KafkaHeaders.TIMESTAMP, Long.class);
-		RecordHeaders recordHeaders = new RecordHeaders();
+		Headers recordHeaders = initialRecordHeaders(message);
 		this.headerMapper.fromHeaders(headers, recordHeaders);
 		return new ProducerRecord(topic == null ? defaultTopic : topic, partition, timestamp, key, payload,
 				recordHeaders);
 	}
 
+	/**
+	 * Subclasses can populate additional headers before they are mapped.
+	 * @param message the message.
+	 * @return the headers
+	 * @since 2.1
+	 */
+	protected Headers initialRecordHeaders(Message<?> message) {
+		return new RecordHeaders();
+	}
+
 	/**
 	 * Subclasses can convert the payload; by default, it's sent unchanged to Kafka.
 	 * @param message the message.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -20,8 +20,11 @@ import java.io.IOException;
 import java.lang.reflect.Type;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
 
 import org.springframework.kafka.support.KafkaNull;
+import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper.TypePrecedence;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
@@ -43,6 +46,8 @@ public class StringJsonMessageConverter extends MessagingMessageConverter {
 
 	private final ObjectMapper objectMapper;
 
+	private Jackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper();
+
 	public StringJsonMessageConverter() {
 		this(new ObjectMapper());
 		this.objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
@@ -54,6 +59,27 @@ public class StringJsonMessageConverter extends MessagingMessageConverter {
 		this.objectMapper = objectMapper;
 	}
 
+	public Jackson2JavaTypeMapper getTypeMapper() {
+		return this.typeMapper;
+	}
+
+	/**
+	 * Set a customized type mapper.
+	 * @param typeMapper the type mapper.
+	 * @since 2.1
+	 */
+	public void setTypeMapper(Jackson2JavaTypeMapper typeMapper) {
+		Assert.notNull(typeMapper, "'typeMapper' cannot be null");
+		this.typeMapper = typeMapper;
+	}
+
+	@Override
+	protected Headers initialRecordHeaders(Message<?> message) {
+		RecordHeaders headers = new RecordHeaders();
+		this.typeMapper.fromClass(message.getPayload().getClass(), headers);
+		return headers;
+	}
+
 	@Override
 	protected Object convertPayload(Message<?> message) {
 		try {
@@ -72,7 +98,12 @@ public class StringJsonMessageConverter extends MessagingMessageConverter {
 			return KafkaNull.INSTANCE;
 		}
 
-		JavaType javaType = TypeFactory.defaultInstance().constructType(type);
+		JavaType javaType = this.typeMapper.getTypePrecedence().equals(TypePrecedence.INFERRED)
+			? TypeFactory.defaultInstance().constructType(type)
+			: this.typeMapper.toJavaType(record.headers());
+		if (javaType == null) { // no headers
+			javaType = TypeFactory.defaultInstance().constructType(type);
+		}
 		if (value instanceof String) {
 			try {
 				return this.objectMapper.readValue((String) value, javaType);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-2017 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.
@@ -21,12 +21,19 @@ import java.util.Arrays;
 import java.util.Map;
 
 import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.ExtendedDeserializer;
 
 import org.springframework.core.ResolvableType;
+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.JavaType;
 import com.fasterxml.jackson.databind.MapperFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectReader;
@@ -38,16 +45,34 @@ import com.fasterxml.jackson.databind.ObjectReader;
  *
  * @author Igor Stepanov
  * @author Artem Bilan
+ * @author Gary Russell
  */
-public class JsonDeserializer<T> implements Deserializer<T> {
+public class JsonDeserializer<T> implements ExtendedDeserializer<T> {
+
+	/**
+	 * Kafka config property for the default key type if no header.
+	 */
+	public static final String DEFAULT_KEY_TYPE = "spring.json.key.default.type";
+
+	/**
+	 * Kafka config property for the default value type if no header.
+	 */
+	public static final String DEFAULT_VALUE_TYPE = "spring.json.default.value.type";
+
+	/**
+	 * Kafka config property for trusted deserialization packages.
+	 */
+	public static final String TRUSTED_PACKAGES = "spring.json.trusted.packages";
 
 	protected final ObjectMapper objectMapper;
 
-	protected final Class<T> targetType;
+	protected Class<T> targetType;
 
 	private volatile ObjectReader reader;
 
-	protected JsonDeserializer() {
+	protected Jackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper();
+
+	public JsonDeserializer() {
 		this((Class<T>) null);
 	}
 
@@ -68,14 +93,89 @@ public class JsonDeserializer<T> implements Deserializer<T> {
 		if (targetType == null) {
 			targetType = (Class<T>) ResolvableType.forClass(getClass()).getSuperType().resolveGeneric(0);
 		}
-		Assert.notNull(targetType, "'targetType' cannot be resolved.");
 		this.targetType = targetType;
 	}
 
-	public void configure(Map<String, ?> configs, boolean isKey) {
-		// No-op
+	public Jackson2JavaTypeMapper getTypeMapper() {
+		return this.typeMapper;
 	}
 
+	/**
+	 * Set a customized type mapper.
+	 * @param typeMapper the type mapper.
+	 * @since 2.1
+	 */
+	public void setTypeMapper(Jackson2JavaTypeMapper typeMapper) {
+		Assert.notNull(typeMapper, "'typeMapper' cannot be null");
+		this.typeMapper = typeMapper;
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public void configure(Map<String, ?> configs, boolean isKey) {
+		try {
+			if (isKey && configs.containsKey(DEFAULT_KEY_TYPE)) {
+				if (configs.get(DEFAULT_KEY_TYPE) instanceof Class) {
+					this.targetType = (Class<T>) configs.get(DEFAULT_KEY_TYPE);
+				}
+				else if (configs.get(DEFAULT_KEY_TYPE) instanceof String) {
+					this.targetType = (Class<T>) ClassUtils.forName((String) configs.get(DEFAULT_KEY_TYPE), null);
+				}
+				else {
+					throw new IllegalStateException(DEFAULT_KEY_TYPE + " must be Class or String");
+				}
+			}
+			else if (!isKey && configs.containsKey(DEFAULT_VALUE_TYPE)) {
+				if (configs.get(DEFAULT_VALUE_TYPE) instanceof Class) {
+					this.targetType = (Class<T>) configs.get(DEFAULT_VALUE_TYPE);
+				}
+				else if (configs.get(DEFAULT_VALUE_TYPE) instanceof String) {
+					this.targetType = (Class<T>) ClassUtils.forName((String) configs.get(DEFAULT_VALUE_TYPE), null);
+				}
+				else {
+					throw new IllegalStateException(DEFAULT_VALUE_TYPE + " must be Class or String");
+				}
+			}
+		}
+		catch (ClassNotFoundException | LinkageError e) {
+			throw new IllegalStateException(e);
+		}
+		if (configs.containsKey(TRUSTED_PACKAGES)) {
+			if (configs.get(TRUSTED_PACKAGES) instanceof String) {
+				this.typeMapper.addTrustedPackages(
+						StringUtils.commaDelimitedListToStringArray((String) configs.get(TRUSTED_PACKAGES)));
+			}
+		}
+	}
+
+	/**
+	 * Add trusted packages for deserialization.
+	 * @param packages the packages.
+	 * @since 2.1
+	 */
+	public void addTrustedPackages(String... packages) {
+		this.typeMapper.addTrustedPackages(packages);
+	}
+
+	@Override
+	public T deserialize(String topic, Headers headers, byte[] data) {
+		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);
+			}
+		}
+	}
+
+	@Override
 	public T deserialize(String topic, byte[] data) {
 		if (this.reader == null) {
 			this.reader = this.objectMapper.readerFor(this.targetType);
@@ -93,6 +193,7 @@ public class JsonDeserializer<T> implements Deserializer<T> {
 		}
 	}
 
+	@Override
 	public void close() {
 		// No-op
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -20,8 +20,12 @@ import java.io.IOException;
 import java.util.Map;
 
 import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.ExtendedSerializer;
 import org.apache.kafka.common.serialization.Serializer;
 
+import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper;
+import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper;
 import org.springframework.util.Assert;
 
 import com.fasterxml.jackson.databind.DeserializationFeature;
@@ -35,11 +39,21 @@ import com.fasterxml.jackson.databind.ObjectMapper;
  *
  * @author Igor Stepanov
  * @author Artem Bilan
+ * @author Gary Russell
  */
-public class JsonSerializer<T> implements Serializer<T> {
+public class JsonSerializer<T> implements ExtendedSerializer<T> {
+
+	/**
+	 * Kafka config property for to disable adding type headers.
+	 */
+	public static final String ADD_TYPE_INFO_HEADERS = "spring.json.add.type.headers";
 
 	protected final ObjectMapper objectMapper;
 
+	protected boolean addTypeInfo = true;
+
+	protected Jackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper();
+
 	public JsonSerializer() {
 		this(new ObjectMapper());
 		this.objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
@@ -51,10 +65,58 @@ public class JsonSerializer<T> implements Serializer<T> {
 		this.objectMapper = objectMapper;
 	}
 
-	public void configure(Map<String, ?> configs, boolean isKey) {
-		// No-op
+	public boolean isAddTypeInfo() {
+		return this.addTypeInfo;
 	}
 
+	/**
+	 * Set to false to disable adding type info headers.
+	 * @param addTypeInfo true to add headers.
+	 * @since 2.1
+	 */
+	public void setAddTypeInfo(boolean addTypeInfo) {
+		this.addTypeInfo = addTypeInfo;
+	}
+
+	public Jackson2JavaTypeMapper getTypeMapper() {
+		return this.typeMapper;
+	}
+
+	/**
+	 * Set a customized type mapper.
+	 * @param typeMapper the type mapper.
+	 * @since 2.1
+	 */
+	public void setTypeMapper(Jackson2JavaTypeMapper typeMapper) {
+		Assert.notNull(typeMapper, "'typeMapper' cannot be null");
+		this.typeMapper = typeMapper;
+	}
+
+	@Override
+	public void configure(Map<String, ?> configs, boolean isKey) {
+		if (configs.containsKey(ADD_TYPE_INFO_HEADERS)) {
+			Object config = configs.get(ADD_TYPE_INFO_HEADERS);
+			if (config instanceof Boolean) {
+				this.addTypeInfo = (Boolean) config;
+			}
+			else if (config instanceof String) {
+				this.addTypeInfo = Boolean.valueOf((String) config);
+			}
+			else {
+				throw new IllegalStateException(ADD_TYPE_INFO_HEADERS + " must be Boolean or String");
+			}
+		}
+	}
+
+	@Override
+	public byte[] serialize(String topic, Headers headers, T data) {
+		if (this.addTypeInfo) {
+			this.typeMapper.fromJavaType(this.objectMapper.constructType(data.getClass()), headers);
+		}
+		return serialize(topic, data);
+	}
+
+	@Override
 	public byte[] serialize(String topic, T data) {
 		try {
 			byte[] result = null;
@@ -68,6 +130,7 @@ public class JsonSerializer<T> implements Serializer<T> {
 		}
 	}
 
+	@Override
 	public void close() {
 		// No-op
 	}

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -79,6 +79,8 @@ import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
+import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper;
+import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper;
 import org.springframework.kafka.support.converter.StringJsonMessageConverter;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -119,7 +121,7 @@ public class EnableKafkaIntegrationTests {
 			"annotated18", "annotated19", "annotated20", "annotated21", "annotated21reply", "annotated22",
 			"annotated22reply", "annotated23", "annotated23reply", "annotated24", "annotated24reply",
 			"annotated25", "annotated25reply1", "annotated25reply2", "annotated26", "annotated27", "annotated28",
-			"annotated29", "annotated30", "annotated30reply");
+			"annotated29", "annotated30", "annotated30reply", "annotated31");
 
 //	@Rule
 //	public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.TRACE,
@@ -310,6 +312,28 @@ public class EnableKafkaIntegrationTests {
 				.isEqualTo("buz.explicitGroupId");
 	}
 
+	@Test
+	@DirtiesContext
+	public void testJsonHeaders() throws Exception {
+		ConcurrentMessageListenerContainer<?, ?> container =
+				(ConcurrentMessageListenerContainer<?, ?>) registry.getListenerContainer("jsonHeaders");
+		Object messageListener = container.getContainerProperties().getMessageListener();
+		DefaultJackson2JavaTypeMapper typeMapper = KafkaTestUtils.getPropertyValue(messageListener,
+				"messageConverter.typeMapper", DefaultJackson2JavaTypeMapper.class);
+		new DirectFieldAccessor(typeMapper).setPropertyValue("typePrecedence",
+				Jackson2JavaTypeMapper.TypePrecedence.TYPE_ID);
+		assertThat(container).isNotNull();
+		Foo foo = new Foo();
+		foo.setBar("bar");
+		this.kafkaJsonTemplate.send(MessageBuilder.withPayload(foo)
+				.setHeader(KafkaHeaders.TOPIC, "annotated31")
+				.setHeader(KafkaHeaders.PARTITION_ID, 0)
+				.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
+				.build());
+		assertThat(this.listener.latch19.await(60, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.foo.getBar()).isEqualTo("bar");
+	}
+
 	@Test
 	public void testNulls() throws Exception {
 		template.send("annotated12", null, null);
@@ -649,7 +673,11 @@ public class EnableKafkaIntegrationTests {
 			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
 					new ConcurrentKafkaListenerContainerFactory<>();
 			factory.setConsumerFactory(consumerFactory());
-			factory.setMessageConverter(new StringJsonMessageConverter());
+			StringJsonMessageConverter converter = new StringJsonMessageConverter();
+			DefaultJackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper();
+			typeMapper.addTrustedPackages("*");
+			converter.setTypeMapper(typeMapper);
+			factory.setMessageConverter(converter);
 			return factory;
 		}
 
@@ -966,6 +994,8 @@ public class EnableKafkaIntegrationTests {
 
 		private final CountDownLatch latch18 = new CountDownLatch(2);
 
+		private final CountDownLatch latch19 = new CountDownLatch(1);
+
 		private final CountDownLatch eventLatch = new CountDownLatch(1);
 
 		private volatile Integer partition;
@@ -1072,6 +1102,14 @@ public class EnableKafkaIntegrationTests {
 			this.latch6.countDown();
 		}
 
+		@KafkaListener(id = "jsonHeaders", topics = "annotated31",
+				containerFactory = "kafkaJsonListenerContainerFactory",
+				groupId = "jsonHeaders")
+		public void jsonHeaders(Bar foo) { // should be mapped to Foo via Headers
+			this.foo = (Foo) foo;
+			this.latch19.countDown();
+		}
+
 		@KafkaListener(id = "rebalanceListener", topics = "annotated11", idIsGroup = false,
 				containerFactory = "kafkaRebalanceListenerContainerFactory")
 		public void listen7(String foo) {
@@ -1313,7 +1351,11 @@ public class EnableKafkaIntegrationTests {
 
 	}
 
-	public static class Foo {
+	public interface Bar {
+
+	}
+
+	public static class Foo implements Bar {
 
 		private String bar;
 

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -52,6 +52,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
 import org.junit.ClassRule;
@@ -75,6 +76,8 @@ import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapte
 import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import org.springframework.kafka.support.serializer.JsonSerializer;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
 import org.springframework.kafka.test.utils.ContainerTestUtils;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -91,6 +94,10 @@ public class KafkaMessageListenerContainerTests {
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
+	private static String topic1 = "testTopic1";
+
+	private static String topic2 = "testTopic2";
+
 	private static String topic3 = "testTopic3";
 
 	private static String topic4 = "testTopic4";
@@ -122,8 +129,9 @@ public class KafkaMessageListenerContainerTests {
 	private static String topic17 = "testTopic17";
 
 	@ClassRule
-	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic3, topic4, topic5,
-			topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16, topic17);
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5,
+			topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16,
+			topic17);
 
 	@Rule
 	public TestName testName = new TestName();
@@ -1435,6 +1443,81 @@ public class KafkaMessageListenerContainerTests {
 		logger.info("Stop manual ack rebalance");
 	}
 
+	@Test
+	public void testJsonSerDeConfiguredType() throws Exception {
+		this.logger.info("Start JSON1");
+		Map<String, Object> props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka);
+		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+		props.put(JsonDeserializer.DEFAULT_VALUE_TYPE, Foo.class);
+		DefaultKafkaConsumerFactory<Integer, Foo> cf = new DefaultKafkaConsumerFactory<>(props);
+		ContainerProperties containerProps = new ContainerProperties(topic1);
+
+		final CountDownLatch latch = new CountDownLatch(1);
+		final AtomicReference<ConsumerRecord<?, ?>> received = new AtomicReference<>();
+		containerProps.setMessageListener((MessageListener<Integer, Foo>) record -> {
+			KafkaMessageListenerContainerTests.this.logger.info("json: " + record);
+			received.set(record);
+			latch.countDown();
+		});
+
+		KafkaMessageListenerContainer<Integer, Foo> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container.setBeanName("testJson1");
+		container.start();
+
+		ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
+
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+		senderProps.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
+		ProducerFactory<Integer, Foo> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, Foo> template = new KafkaTemplate<>(pf);
+		template.setDefaultTopic(topic1);
+		template.sendDefault(0, new Foo("bar"));
+		template.flush();
+		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
+		assertThat(received.get().value()).isInstanceOf(Foo.class);
+		container.stop();
+		this.logger.info("Stop JSON1");
+	}
+
+	@Test
+	public void testJsonSerDeHeaderSimpleType() throws Exception {
+		this.logger.info("Start JSON2");
+		Map<String, Object> props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka);
+		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+		props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
+		DefaultKafkaConsumerFactory<Integer, Foo> cf = new DefaultKafkaConsumerFactory<>(props);
+		ContainerProperties containerProps = new ContainerProperties(topic2);
+
+		final CountDownLatch latch = new CountDownLatch(1);
+		final AtomicReference<ConsumerRecord<?, ?>> received = new AtomicReference<>();
+		containerProps.setMessageListener((MessageListener<Integer, Foo>) record -> {
+			KafkaMessageListenerContainerTests.this.logger.info("json: " + record);
+			received.set(record);
+			latch.countDown();
+		});
+
+		KafkaMessageListenerContainer<Integer, Foo> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container.setBeanName("testJson2");
+		container.start();
+
+		ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
+
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+		ProducerFactory<Integer, Foo> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, Foo> template = new KafkaTemplate<>(pf);
+		template.setDefaultTopic(topic2);
+		template.sendDefault(0, new Foo("bar"));
+		template.flush();
+		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
+		assertThat(received.get().value()).isInstanceOf(Foo.class);
+		container.stop();
+		this.logger.info("Stop JSON2");
+	}
+
 	private Consumer<?, ?> spyOnConsumer(KafkaMessageListenerContainer<Integer, String> container) {
 		Consumer<?, ?> consumer = spy(
 				KafkaTestUtils.getPropertyValue(container, "listenerConsumer.consumer", Consumer.class));
@@ -1465,4 +1548,31 @@ public class KafkaMessageListenerContainerTests {
 
 	}
 
+	public static class Foo {
+
+		private String bar;
+
+		public Foo() {
+			super();
+		}
+
+		public Foo(String bar) {
+			this.bar = bar;
+		}
+
+		public String getBar() {
+			return this.bar;
+		}
+
+		public void setBar(String bar) {
+			this.bar = bar;
+		}
+
+		@Override
+		public String toString() {
+			return "Foo [bar=" + this.bar + "]";
+		}
+
+	}
+
 }

src/reference/asciidoc/appendix.adoc

@@ -1,4 +1,4 @@
 [[history]]
 == Change History
 
-include::./changes-1.0-1.1.adoc[]
+include::./changes-since-1.0.adoc[]

src/reference/asciidoc/changes-since-1.0.adoc

@@ -1,4 +1,30 @@
 [[migration]]
+=== Changes Between 1.3 an 2.0
+
+==== Spring Framework and Java Versions
+
+The Spring for Apache Kafka project now requires Spring Framework 5.0 and Java 8.
+
+==== @KafkaListener Changes
+
+You can now annotate `@KafkaListener` methods (and classes, and `@KafkaHandler` methods) with `@SendTo`.
+If the method returns a result, it is forwarded to the specified topic.
+See <<annotation-send-to>> for more information.
+
+==== Message Listeners
+
+Message listeners can now be aware of the `Consumer` object.
+See <<message-listeners>> for more information.
+
+==== ConsumerAwareRebalanceListener
+
+Rebalance listeners can now access the `Consumer` object during rebalance notifications.
+See <<rebalance-listeners>> for more information.
+
+==== @EmbeddedKafka Annotation
+
+For convenience a test class level `@EmbeddedKafka` annotation is provided with the purpose to register `KafkaEmbedded` as a bean.
+See <<testing>> for more information.
 === Changes Between 1.2 and 1.3
 
 ==== Support for Transactions

src/reference/asciidoc/kafka.adoc

@@ -1186,6 +1186,14 @@ Both `JsonSerializer` and `JsonDeserializer` can be customized with an `ObjectMa
 You can also extend them to implement some particular configuration logic in the
 `configure(Map<String, ?> configs, boolean isKey)` method.
 
+Starting with _version 2.1_, type information can be conveyed in record `Headers`, allowing the handling of multiple types.
+In addition, the serializer/deserializer can be configured using Kafka properties.
+
+-  `JsonSerializer.ADD_TYPE_INFO_HEADERS` (default `true`); set to `false` to disable this feature.
+- `JsonDeserializer.DEFAULT_KEY_TYPE`; fallback type for deserialization of keys if no header information is present.
+- `JsonDeserializer.DEFAULT_VALUE_TYPE`; fallback type for deserialization of values if no header information is present.
+- `JsonDeserializer.TRUSTED_PACKAGES` (default `java.util`, `java.lang`); comma-delimited list of package patterns allowed for deserialization; `*` means deserialize all.
+
 Although the `Serializer`/`Deserializer` API is quite simple and flexible from the low-level Kafka `Consumer` and
 `Producer` perspective, you might need more flexibility at the Spring Messaging level, either when using `@KafkaListener` or <<si-kafka,Spring Integration>>.
 To easily convert to/from `org.springframework.messaging.Message`, Spring for Apache Kafka provides a `MessageConverter`
@@ -1223,6 +1231,11 @@ NOTE: When using the `StringJsonMessageConverter`, you should use a `StringDeser
 
 Starting with _version 1.3.2_ you can also use a `StringJsonMessageConverter` within a `BatchMessagingMessageConverter` for converting batch messages, when using a batch listener container factory.
 
+By default, the type for the conversion is inferred from the listener argument.
+If you configure the `StringJsonMessageConverter` with a `DefaultJackson2TypeMapper` that has its `TypePrecedence` set to `TYPE_ID` (instead of the default `INFERRED`), then the converter will use type information in headers (if present) instead.
+This allows, for example, listener methods to be declared with interfaces instead of concrete classes.
+Also, the type converter supports mapping so the deserialization can be to a different type than the source (as long as the data is compatible).
+
 [source, java]
 ----
 @Bean

src/reference/asciidoc/whats-new.adoc

@@ -1,26 +1,10 @@
-=== What's new in 2.0 Since 1.3
+=== What's new in 2.1 Since 2.0
 
-==== Spring Framework and Java Versions
+==== Kafka Client Version
 
-The Spring for Apache Kafka project now requires Spring Framework 5.0 and Java 8.
+This version requires the 1.0.0 `kafka-clients` or higher.
 
-==== @KafkaListener Changes
+==== JSON Improvements
 
-You can now annotate `@KafkaListener` methods (and classes, and `@KafkaHandler` methods) with `@SendTo`.
-If the method returns a result, it is forwarded to the specified topic.
-See <<annotation-send-to>> for more information.
-
-==== Message Listeners
-
-Message listeners can now be aware of the `Consumer` object.
-See <<message-listeners>> for more information.
-
-==== ConsumerAwareRebalanceListener
-
-Rebalance listeners can now access the `Consumer` object during rebalance notifications.
-See <<rebalance-listeners>> for more information.
-
-==== @EmbeddedKafka Annotation
-
-For convenience a test class level `@EmbeddedKafka` annotation is provided with the purpose to register `KafkaEmbedded` as a bean.
-See <<testing>> for more information.
+The `StringJsonMessageConverter` and `JsonSerializer` now add type information in `Headers`, allowing the converter and `JsonDeserializer` to create specific types on reception, based on the message itself rather than a fixed configured type.
+See <<serdes>> for more information.