GH-794: Json Deserializer - remove type headers

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

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

@@ -182,4 +182,19 @@ public class DefaultJackson2JavaTypeMapper extends AbstractJavaTypeMapper
 		return toJavaType(headers).getRawClass();
 	}
 
+	@Override
+	public void removeHeaders(Headers headers) {
+		try {
+			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
+		}
+	}
+
 }

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

@@ -68,4 +68,13 @@ public interface Jackson2JavaTypeMapper extends ClassMapper {
 
 	void addTrustedPackages(String... packages);
 
+	/**
+	 * Remove the type information headers.
+	 * @param headers the headers.
+	 * @since 2.2
+	 */
+	default void removeHeaders(Headers headers) {
+		// NOSONAR
+	}
+
 }

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

@@ -89,6 +89,11 @@ public class JsonDeserializer<T> implements ExtendedDeserializer<T> {
 	 */
 	public static final String TYPE_MAPPINGS = JsonSerializer.TYPE_MAPPINGS;
 
+	/**
+	 * Kafka config property for removing type headers.
+	 */
+	public static final String REMOVE_TYPE_INFO_HEADERS = "spring.json.remove.type.headers";
+
 	protected final ObjectMapper objectMapper;
 
 	protected Class<T> targetType;
@@ -99,6 +104,8 @@ public class JsonDeserializer<T> implements ExtendedDeserializer<T> {
 
 	private boolean typeMapperExplicitlySet = false;
 
+	private boolean removeTypeHeaders = true;
+
 	/**
 	 * Construct an instance with a default {@link ObjectMapper}.
 	 */
@@ -203,6 +210,16 @@ public class JsonDeserializer<T> implements ExtendedDeserializer<T> {
 		}
 	}
 
+	/**
+	 * Set to false to retain type information headers after deserialization.
+	 * Default true.
+	 * @param removeTypeHeaders true to remove headers.
+	 * @since 2.1
+	 */
+	public void setRemoveTypeHeaders(boolean removeTypeHeaders) {
+		this.removeTypeHeaders = removeTypeHeaders;
+	}
+
 	@SuppressWarnings("unchecked")
 	@Override
 	public void configure(Map<String, ?> configs, boolean isKey) {
@@ -263,6 +280,9 @@ public class JsonDeserializer<T> implements ExtendedDeserializer<T> {
 			((AbstractJavaTypeMapper) this.typeMapper).setIdClassMapping(
 					JsonSerializer.createMappings((String) configs.get(JsonSerializer.TYPE_MAPPINGS)));
 		}
+		if (configs.containsKey(REMOVE_TYPE_INFO_HEADERS)) {
+			this.removeTypeHeaders = Boolean.parseBoolean((String) configs.get(REMOVE_TYPE_INFO_HEADERS));
+		}
 	}
 
 	/**
@@ -292,6 +312,9 @@ public class JsonDeserializer<T> implements ExtendedDeserializer<T> {
 				reader = this.objectMapper.readerFor(javaType);
 			}
 		}
+		if (this.removeTypeHeaders) {
+			this.typeMapper.removeHeaders(headers);
+		}
 		if (reader == null) {
 			reader = this.reader;
 		}

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

@@ -1647,6 +1647,7 @@ public class KafkaMessageListenerContainerTests {
 		assertThat(received.get().value()).isInstanceOf(Foo.class);
 		container.stop();
 		this.logger.info("Stop JSON2");
+		assertThat(received.get().headers().iterator().hasNext()).isFalse();
 	}
 
 	@Test

src/reference/asciidoc/kafka.adoc

@@ -1629,11 +1629,15 @@ 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.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.
 - `JsonDeserializer.TRUSTED_PACKAGES` (default `java.util`, `java.lang`); comma-delimited list of package patterns allowed for deserialization; `*` means deserialize all.
 - `JsonDeserializer.TYPE_MAPPINGS` (default `empty`); see below.
 
+Starting with version 2.2, the type information headers (if added by the serializer) will be removed by the deserializer.
+You can revert to the previous behavior by setting the `removeTypeHeaders` property to false, either directly on the deserializer, or with the configuration property described above.
+
 **Mapping Types**
 
 Starting with version 2.2, you can now provide type mappings using the properties in the above list; previously you had to customize the type mapper within the serializer, deserializer.

src/reference/asciidoc/whats-new.adoc

@@ -70,6 +70,8 @@ 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.
+
 See <<serdes>> for more information.
 
 ==== Kafka Streams Changes