GH-1137: Fix Json Type Headers for KStream

Fixes #1137

- use `lastHeader()` to ensure we retrieve the latest header
- clear the headers when serializing, if already present
- also expose the existing deser `useTypeHeaders` Kafka property as a property directly

**cherry-pick to 2.2.x**

Author: garyrussell

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

@@ -19,7 +19,6 @@
 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;
@@ -157,12 +156,11 @@ protected String retrieveHeader(Headers headers, String headerName) {
 	}
 
 	protected String retrieveHeaderAsString(Headers headers, String headerName) {
-		Iterator<Header> headerValues = headers.headers(headerName).iterator();
-		if (headerValues.hasNext()) {
-			Header headerValue = headerValues.next();
+		Header header = headers.lastHeader(headerName);
+		if (header != null) {
 			String classId = null;
-			if (headerValue.value() != null) {
-				classId = new String(headerValue.value(), StandardCharsets.UTF_8);
+			if (header.value() != null) {
+				classId = new String(header.value(), StandardCharsets.UTF_8);
 			}
 			return classId;
 		}

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

@@ -160,7 +160,12 @@ private boolean isTrustedPackage(String requestedType) {
 
 	@Override
 	public void fromJavaType(JavaType javaType, Headers headers) {
-		addHeader(headers, getClassIdFieldName(), javaType.getRawClass());
+		String classIdFieldName = getClassIdFieldName();
+		if (headers.lastHeader(classIdFieldName) != null) {
+			removeHeaders(headers);
+		}
+
+		addHeader(headers, classIdFieldName, javaType.getRawClass());
 
 		if (javaType.isContainerType() && !javaType.isArrayType()) {
 			addHeader(headers, getContentClassIdFieldName(), javaType.getContentType().getRawClass());

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

@@ -18,6 +18,7 @@
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Map;
 import java.util.function.Consumer;
 
@@ -119,6 +120,8 @@ public class JsonDeserializer<T> implements ExtendedDeserializer<T> {
 
 	private boolean removeTypeHeaders = true;
 
+	private boolean useTypeHeaders = true;
+
 	/**
 	 * Construct an instance with a default {@link ObjectMapper}.
 	 */
@@ -239,6 +242,21 @@ public void setRemoveTypeHeaders(boolean removeTypeHeaders) {
 		this.removeTypeHeaders = removeTypeHeaders;
 	}
 
+	/**
+	 * Set to false to ignore type information in headers and use the configured
+	 * target type instead.
+	 * Only applies if the preconfigured type mapper is used.
+	 * Default true.
+	 * @param useTypeHeaders false to ignore type headers.
+	 * @since 2.2.8
+	 */
+	public void setUseTypeHeaders(boolean useTypeHeaders) {
+		if (!this.typeMapperExplicitlySet) {
+			this.useTypeHeaders = useTypeHeaders;
+			setUpTypePrecedence(Collections.emptyMap());
+		}
+	}
+
 	@Override
 	public void configure(Map<String, ?> configs, boolean isKey) {
 		setUseTypeMapperForKey(isKey);
@@ -261,11 +279,10 @@ public void configure(Map<String, ?> configs, boolean isKey) {
 
 	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.useTypeHeaders = Boolean.parseBoolean(configs.get(USE_TYPE_INFO_HEADERS).toString());
 			}
-			this.typeMapper.setTypePrecedence(useTypeHeaders ? TypePrecedence.TYPE_ID : TypePrecedence.INFERRED);
+			this.typeMapper.setTypePrecedence(this.useTypeHeaders ? TypePrecedence.TYPE_ID : TypePrecedence.INFERRED);
 		}
 	}
 

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

@@ -195,6 +195,9 @@ public void testDeserTypeHeadersConfig() {
 		assertThat(KafkaTestUtils.getPropertyValue(this.jsonReader, "typeMapper.typePrecedence"))
 			.isEqualTo(TypePrecedence.TYPE_ID);
 		this.jsonReader.configure(Collections.singletonMap(JsonDeserializer.USE_TYPE_INFO_HEADERS, false), false);
+		assertThat(KafkaTestUtils.getPropertyValue(this.jsonReader, "typeMapper.typePrecedence"))
+			.isEqualTo(TypePrecedence.INFERRED);
+		this.jsonReader.setUseTypeHeaders(true);
 		this.jsonReader.configure(Collections.emptyMap(), false);
 		assertThat(KafkaTestUtils.getPropertyValue(this.jsonReader, "typeMapper.typePrecedence"))
 			.isEqualTo(TypePrecedence.TYPE_ID);
@@ -212,6 +215,36 @@ public void testDeserializerTypeInference() {
 		assertThat(de.deserialize(topic, ser.serialize(topic, dummy))).isEqualTo(dummy);
 	}
 
+	@Test
+	public void testPreExistingHeaders() {
+		JsonSerializer<? super Foo> ser = new JsonSerializer<>();
+		Headers headers = new RecordHeaders();
+		ser.serialize("", headers, new Foo());
+		byte[] data = ser.serialize("", headers, new Bar());
+		JsonDeserializer<? super Foo> deser = new JsonDeserializer<>();
+		deser.setRemoveTypeHeaders(false);
+		deser.addTrustedPackages(this.getClass().getPackage().getName());
+		assertThat(deser.deserialize("", headers, data)).isInstanceOf(Bar.class);
+		assertThat(headers.headers(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME)).hasSize(1);
+		ser.close();
+		deser.close();
+	}
+
+	@Test
+	public void testDontUseTypeHeaders() {
+		JsonSerializer<? super Foo> ser = new JsonSerializer<>();
+		Headers headers = new RecordHeaders();
+		byte[] data = ser.serialize("", headers, new Bar());
+		JsonDeserializer<? super Foo> deser = new JsonDeserializer<>(Foo.class);
+		deser.setRemoveTypeHeaders(false);
+		deser.setUseTypeHeaders(false);
+		deser.addTrustedPackages(this.getClass().getPackage().getName());
+		assertThat(deser.deserialize("", headers, data)).isExactlyInstanceOf(Foo.class);
+		assertThat(headers.headers(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME)).hasSize(1);
+		ser.close();
+		deser.close();
+	}
+
 	static class DummyEntityJsonDeserializer extends JsonDeserializer<DummyEntity> {
 
 	}
@@ -220,4 +253,16 @@ static class DummyEntityArrayJsonDeserializer extends JsonDeserializer<DummyEnti
 
 	}
 
+	public static class Foo {
+
+		public String foo = "foo";
+
+	}
+
+	public static class Bar extends Foo {
+
+		public String bar = "bar";
+
+	}
+
 }