Improve JsonDeserializer to keep compatibility

Since version 2.1.0, spring-kafka add whitelist of trusted packages for
security, It will break existing application, It must manual add generic
type's package like
this.addTrustedPackages(Trade.class.getPackage().getName()), We should
add Trade's package as default for class TradeDeserializer extends
JsonDeserializer<Trade>.

Polishing

- also add package if targetType comes from Kafka consumer properties.

Author: quaff

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

@@ -46,6 +46,7 @@
  * @author Igor Stepanov
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Yanming Zhou
  */
 public class JsonDeserializer<T> implements ExtendedDeserializer<T> {
 
@@ -94,6 +95,7 @@ public JsonDeserializer(Class<T> targetType, ObjectMapper objectMapper) {
 			targetType = (Class<T>) ResolvableType.forClass(getClass()).getSuperType().resolveGeneric(0);
 		}
 		this.targetType = targetType;
+		addTargetPackageToTrusted();
 	}
 
 	public Jackson2JavaTypeMapper getTypeMapper() {
@@ -135,6 +137,7 @@ else if (configs.get(DEFAULT_VALUE_TYPE) instanceof String) {
 				else {
 					throw new IllegalStateException(DEFAULT_VALUE_TYPE + " must be Class or String");
 				}
+				addTargetPackageToTrusted();
 			}
 		}
 		catch (ClassNotFoundException | LinkageError e) {
@@ -157,6 +160,12 @@ public void addTrustedPackages(String... packages) {
 		this.typeMapper.addTrustedPackages(packages);
 	}
 
+	private void addTargetPackageToTrusted() {
+		if (this.targetType != null) {
+			addTrustedPackages(this.targetType.getPackage().getName());
+		}
+	}
+
 	@Override
 	public T deserialize(String topic, Headers headers, byte[] data) {
 		JavaType javaType = this.typeMapper.toJavaType(headers);

spring-kafka/src/test/java/org/springframework/kafka/support/serializer/JsonSerializationTests.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.
@@ -24,18 +24,22 @@
 import java.util.List;
 
 import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.junit.Before;
 import org.junit.Test;
 
+import org.springframework.kafka.support.converter.AbstractJavaTypeMapper;
 import org.springframework.kafka.support.serializer.testentities.DummyEntity;
 
 import com.fasterxml.jackson.core.JsonParseException;
 
 /**
  * @author Igor Stepanov
  * @author Artem Bilan
+ * @author Yanming Zhou
  */
 public class JsonSerializationTests {
 
@@ -47,6 +51,8 @@ public class JsonSerializationTests {
 
 	private JsonDeserializer<DummyEntity> jsonReader;
 
+	private JsonDeserializer<DummyEntity> dummyEntityJsonDeserializer;
+
 	private DummyEntity entity;
 
 	private String topic;
@@ -73,6 +79,7 @@ public void init() {
 		stringReader.configure(new HashMap<String, Object>(), false);
 		stringWriter = new StringSerializer();
 		stringWriter.configure(new HashMap<String, Object>(), false);
+		dummyEntityJsonDeserializer = new DummyEntityJsonDeserializer();
 	}
 
 	/*
@@ -83,6 +90,9 @@ public void init() {
 	@Test
 	public void testDeserializeSerializedEntityEquals() {
 		assertThat(jsonReader.deserialize(topic, jsonWriter.serialize(topic, entity))).isEqualTo(entity);
+		Headers headers = new RecordHeaders();
+		headers.add(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME, DummyEntity.class.getName().getBytes());
+		assertThat(dummyEntityJsonDeserializer.deserialize(topic, headers, jsonWriter.serialize(topic, entity))).isEqualTo(entity);
 	}
 
 	/*
@@ -103,6 +113,18 @@ public void testDeserializeSerializedDummyException() {
 		catch (Exception e) {
 			fail("Expected SerializationException, not " + e.getClass());
 		}
+		try {
+			Headers headers = new RecordHeaders();
+			headers.add(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME, "com.malware.DummyEntity".getBytes());
+			dummyEntityJsonDeserializer.deserialize(topic, headers, jsonWriter.serialize(topic, entity));
+			fail("Expected IllegalArgumentException");
+		}
+		catch (IllegalArgumentException e) {
+			assertThat(e.getMessage()).contains("not in the trusted packages");
+		}
+		catch (Exception e) {
+			fail("Expected IllegalArgumentException, not " + e.getClass());
+		}
 	}
 
 	@Test
@@ -125,4 +147,8 @@ public void testDeserializedJsonNullEqualsNull() {
 		assertThat(jsonReader.deserialize(topic, null)).isEqualTo(null);
 	}
 
+	static class DummyEntityJsonDeserializer extends JsonDeserializer<DummyEntity> {
+
+	}
+
 }