JsonDeserializer Header Removal Polishing

Only remove the headers for the object being deserialized.

Previously, it removed all type headers, with explicit reference to the key
headers. This happened to work ok because the key was deserialized first.

If this ever changes, the key headers would be removed when the value is
deserialized and would not be available for key deserialization.

* `JsonDeserializer`: add `spring.json.use.type.headers` configuration property

Author: garyrussell

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

@@ -188,9 +188,6 @@ public void removeHeaders(Headers headers) {
 			headers.remove(getClassIdFieldName());
 			headers.remove(getContentClassIdFieldName());
 			headers.remove(getKeyClassIdFieldName());
-			headers.remove(KEY_DEFAULT_CLASSID_FIELD_NAME);
-			headers.remove(KEY_DEFAULT_CONTENT_CLASSID_FIELD_NAME);
-			headers.remove(KEY_DEFAULT_KEY_CLASSID_FIELD_NAME);
 		}
 		catch (Exception e) { // NOSONAR
 			// NOSONAR

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

@@ -96,10 +96,16 @@ public class JsonDeserializer<T> implements ExtendedDeserializer<T> {
 	public static final String TYPE_MAPPINGS = JsonSerializer.TYPE_MAPPINGS;
 
 	/**
-	 * Kafka config property for removing type headers.
+	 * Kafka config property for removing type headers (default true).
 	 */
 	public static final String REMOVE_TYPE_INFO_HEADERS = "spring.json.remove.type.headers";
 
+	/**
+	 * Kafka config property for using type headers (default true).
+	 * @since 2.2.3
+	 */
+	public static final String USE_TYPE_INFO_HEADERS = "spring.json.use.type.headers";
+
 	protected final ObjectMapper objectMapper; // NOSONAR
 
 	protected Class<T> targetType; // NOSONAR
@@ -235,6 +241,7 @@ public void setRemoveTypeHeaders(boolean removeTypeHeaders) {
 	@Override
 	public void configure(Map<String, ?> configs, boolean isKey) {
 		setUseTypeMapperForKey(isKey);
+		setUpTypePrecedence(configs);
 		setupTarget(configs, isKey);
 		if (configs.containsKey(TRUSTED_PACKAGES)
 				&& configs.get(TRUSTED_PACKAGES) instanceof String) {
@@ -244,10 +251,20 @@ public void configure(Map<String, ?> configs, boolean isKey) {
 		if (configs.containsKey(TYPE_MAPPINGS) && !this.typeMapperExplicitlySet
 				&& this.typeMapper instanceof AbstractJavaTypeMapper) {
 			((AbstractJavaTypeMapper) this.typeMapper).setIdClassMapping(
-					JsonSerializer.createMappings((String) configs.get(JsonSerializer.TYPE_MAPPINGS)));
+					JsonSerializer.createMappings(configs.get(JsonSerializer.TYPE_MAPPINGS).toString()));
 		}
 		if (configs.containsKey(REMOVE_TYPE_INFO_HEADERS)) {
-			this.removeTypeHeaders = Boolean.parseBoolean((String) configs.get(REMOVE_TYPE_INFO_HEADERS));
+			this.removeTypeHeaders = Boolean.parseBoolean(configs.get(REMOVE_TYPE_INFO_HEADERS).toString());
+		}
+	}
+
+	private void setUpTypePrecedence(Map<String, ?> configs) {
+		if (!this.typeMapperExplicitlySet) {
+			boolean useTypeHeaders = true;
+			if (configs.containsKey(USE_TYPE_INFO_HEADERS)) {
+				useTypeHeaders = Boolean.parseBoolean(configs.get(USE_TYPE_INFO_HEADERS).toString());
+			}
+			this.typeMapper.setTypePrecedence(useTypeHeaders ? TypePrecedence.TYPE_ID : TypePrecedence.INFERRED);
 		}
 	}
 

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

@@ -1638,9 +1638,10 @@ public void testJsonSerDeConfiguredType() throws Exception {
 	public void testJsonSerDeHeaderSimpleType() throws Exception {
 		this.logger.info("Start JSON2");
 		Map<String, Object> props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka);
+		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
 		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
 		props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
-		DefaultKafkaConsumerFactory<Integer, Foo> cf = new DefaultKafkaConsumerFactory<>(props);
+		DefaultKafkaConsumerFactory<Bar, Foo> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic2);
 
 		final CountDownLatch latch = new CountDownLatch(1);
@@ -1651,21 +1652,23 @@ public void testJsonSerDeHeaderSimpleType() throws Exception {
 			latch.countDown();
 		});
 
-		KafkaMessageListenerContainer<Integer, Foo> container =
+		KafkaMessageListenerContainer<Bar, 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.KEY_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
 		senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
-		ProducerFactory<Integer, Foo> pf = new DefaultKafkaProducerFactory<>(senderProps);
-		KafkaTemplate<Integer, Foo> template = new KafkaTemplate<>(pf);
+		ProducerFactory<Bar, Foo> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Bar, Foo> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(topic2);
-		template.sendDefault(0, new Foo("bar"));
+		template.sendDefault(new Bar("foo"), new Foo("bar"));
 		template.flush();
 		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
+		assertThat(received.get().key()).isInstanceOf(Bar.class);
 		assertThat(received.get().value()).isInstanceOf(Foo.class);
 		container.stop();
 		this.logger.info("Stop JSON2");

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

@@ -20,6 +20,7 @@
 import static org.junit.Assert.fail;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 
@@ -32,7 +33,10 @@
 import org.junit.Test;
 
 import org.springframework.kafka.support.converter.AbstractJavaTypeMapper;
+import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper;
+import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper.TypePrecedence;
 import org.springframework.kafka.support.serializer.testentities.DummyEntity;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
 
 import com.fasterxml.jackson.core.JsonParseException;
 
@@ -41,6 +45,7 @@
  * @author Artem Bilan
  * @author Yanming Zhou
  * @author Torsten Schleede
+ * @author Gary Russell
  */
 public class JsonSerializationTests {
 
@@ -180,6 +185,24 @@ public void testExtraFieldIgnored() {
 		deser.close();
 	}
 
+	@Test
+	public void testDeserTypeHeadersConfig() {
+		this.jsonReader.configure(Collections.singletonMap(JsonDeserializer.USE_TYPE_INFO_HEADERS, false), false);
+		assertThat(KafkaTestUtils.getPropertyValue(this.jsonReader, "typeMapper.typePrecedence"))
+			.isEqualTo(TypePrecedence.INFERRED);
+		this.jsonReader.configure(Collections.singletonMap(JsonDeserializer.USE_TYPE_INFO_HEADERS, true), false);
+		assertThat(KafkaTestUtils.getPropertyValue(this.jsonReader, "typeMapper.typePrecedence"))
+			.isEqualTo(TypePrecedence.TYPE_ID);
+		this.jsonReader.configure(Collections.singletonMap(JsonDeserializer.USE_TYPE_INFO_HEADERS, false), false);
+		this.jsonReader.configure(Collections.emptyMap(), false);
+		assertThat(KafkaTestUtils.getPropertyValue(this.jsonReader, "typeMapper.typePrecedence"))
+			.isEqualTo(TypePrecedence.TYPE_ID);
+		this.jsonReader.setTypeMapper(new DefaultJackson2JavaTypeMapper());
+		this.jsonReader.configure(Collections.singletonMap(JsonDeserializer.USE_TYPE_INFO_HEADERS, true), false);
+		assertThat(KafkaTestUtils.getPropertyValue(this.jsonReader, "typeMapper.typePrecedence"))
+			.isEqualTo(TypePrecedence.INFERRED);
+	}
+
 	static class DummyEntityJsonDeserializer extends JsonDeserializer<DummyEntity> {
 
 	}

src/reference/asciidoc/kafka.adoc

@@ -1877,6 +1877,7 @@ In addition, the serializer/deserializer can be configured using Kafka propertie
 
 - `JsonSerializer.ADD_TYPE_INFO_HEADERS` (default `true`); set to `false` to disable this feature on the `JsonSerializer` (sets the `addTypeInfo` property).
 - `JsonSerializer.TYPE_MAPPINGS` (default `empty`); see below.
+- `JsonDeserializer.USE_TYPE_INFO_HEADERS` (default `true`); set to `false` to ignore headers set by the serializer.
 - `JsonDeserializer.REMOVE_TYPE_INFO_HEADERS` (default `true`); set to `false` to retain headers set by the serializer.
 - `JsonDeserializer.KEY_DEFAULT_TYPE`; fallback type for deserialization of keys if no header information is present.
 - `JsonDeserializer.VALUE_DEFAULT_TYPE`; fallback type for deserialization of values if no header information is present.

src/reference/asciidoc/whats-new.adoc

@@ -81,7 +81,9 @@ You can now provide type mapping information using producer/consumer properties.
 
 New constructors are available on the deserializer to allow overriding the type header information with the supplied target type.
 
-The JsonDeserializer will now remove any type information headers by default.
+The `JsonDeserializer` will now remove any type information headers by default.
+
+The `JsonDeserializer` can now be configured to ignore type information headers using a kafka property (since 2.2.3).
 
 See <<serdes>> for more information.