Json (de)serialization configuration improvements (#1573)

* Allow passing JavaType to JsonSerde

* Allow setting forced type for JsonSerializer

* Add copying json serializers with new type

Author: nivertius

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

@@ -142,7 +142,7 @@ public JsonDeserializer(ObjectMapper objectMapper) {
 	 * {@link ObjectMapper}.
 	 * @param targetType the target type to use if no type info headers are present.
 	 */
-	public JsonDeserializer(Class<? super T> targetType) {
+	public JsonDeserializer(@Nullable Class<? super T> targetType) {
 		this(targetType, true);
 	}
 
@@ -151,7 +151,17 @@ public JsonDeserializer(Class<? super T> targetType) {
 	 * @param targetType the target type reference to use if no type info headers are present.
 	 * @since 2.3
 	 */
-	public JsonDeserializer(TypeReference<? super T> targetType) {
+	public JsonDeserializer(@Nullable TypeReference<? super T> targetType) {
+		this(targetType, true);
+	}
+
+
+	/**
+	 * Construct an instance with the provided target type, and a default {@link ObjectMapper}.
+	 * @param targetType the target java type to use if no type info headers are present.
+	 * @since 2.3
+	 */
+	public JsonDeserializer(@Nullable JavaType targetType) {
 		this(targetType, true);
 	}
 
@@ -179,6 +189,18 @@ public JsonDeserializer(TypeReference<? super T> targetType, boolean useHeadersI
 		this(targetType, JacksonUtils.enhancedObjectMapper(), useHeadersIfPresent);
 	}
 
+	/**
+	 * Construct an instance with the provided target type, and
+	 * useHeadersIfPresent with a default {@link ObjectMapper}.
+	 * @param targetType the target java type.
+	 * @param useHeadersIfPresent true to use headers if present and fall back to target
+	 * type if not.
+	 * @since 2.3
+	 */
+	public JsonDeserializer(JavaType targetType, boolean useHeadersIfPresent) {
+		this(targetType, JacksonUtils.enhancedObjectMapper(), useHeadersIfPresent);
+	}
+
 	/**
 	 * Construct an instance with the provided target type, and {@link ObjectMapper}.
 	 * @param targetType the target type to use if no type info headers are present.
@@ -197,6 +219,15 @@ public JsonDeserializer(TypeReference<? super T> targetType, ObjectMapper object
 		this(targetType, objectMapper, true);
 	}
 
+	/**
+	 * Construct an instance with the provided target type, and {@link ObjectMapper}.
+	 * @param targetType the target java type to use if no type info headers are present.
+	 * @param objectMapper the mapper. type if not.
+	 */
+	public JsonDeserializer(JavaType targetType, ObjectMapper objectMapper) {
+		this(targetType, objectMapper, true);
+	}
+
 	/**
 	 * Construct an instance with the provided target type, {@link ObjectMapper} and
 	 * useHeadersIfPresent.
@@ -517,6 +548,43 @@ public void close() {
 		// No-op
 	}
 
+	/**
+	 * Copies this deserializer with same configuration, except new target type is used.
+	 * @param newTargetType type used for when type headers are missing, not null
+	 * @param <X> new deserialization result type
+	 * @return new instance of deserializer with type changes
+	 * @since 2.6
+	 */
+	public <X> JsonDeserializer<X> copyWithType(Class<? super X> newTargetType) {
+		return copyWithType(this.objectMapper.constructType(newTargetType));
+	}
+
+	/**
+	 * Copies this deserializer with same configuration, except new target type reference is used.
+	 * @param newTargetType type reference used for when type headers are missing, not null
+	 * @param <X> new deserialization result type
+	 * @return new instance of deserializer with type changes
+	 * @since 2.6
+	 */
+	public <X> JsonDeserializer<X> copyWithType(TypeReference<? super X> newTargetType) {
+		return copyWithType(this.objectMapper.constructType(newTargetType.getType()));
+	}
+
+	/**
+	 * Copies this deserializer with same configuration, except new target java type is used.
+	 * @param newTargetType java type used for when type headers are missing, not null
+	 * @param <X> new deserialization result type
+	 * @return new instance of deserializer with type changes
+	 * @since 2.6
+	 */
+	public <X> JsonDeserializer<X> copyWithType(JavaType newTargetType) {
+		JsonDeserializer<X> result = new JsonDeserializer<>(newTargetType, this.objectMapper, this.useTypeHeaders);
+		result.removeTypeHeaders = this.removeTypeHeaders;
+		result.typeMapper = this.typeMapper;
+		result.typeMapperExplicitlySet = this.typeMapperExplicitlySet;
+		return result;
+	}
+
 	// Fluent API
 
 	/**

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

@@ -28,6 +28,8 @@
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 /**
@@ -53,29 +55,45 @@ public class JsonSerde<T> implements Serde<T> {
 	private final JsonDeserializer<T> jsonDeserializer;
 
 	public JsonSerde() {
-		this((ObjectMapper) null);
+		this((JavaType) null, JacksonUtils.enhancedObjectMapper());
 	}
 
-	public JsonSerde(Class<? super T> targetType) {
-		this(targetType, null);
+	public JsonSerde(@Nullable Class<? super T> targetType) {
+		this(targetType, JacksonUtils.enhancedObjectMapper());
+	}
+
+	public JsonSerde(@Nullable TypeReference<? super T> targetType) {
+		this(targetType, JacksonUtils.enhancedObjectMapper());
+	}
+
+	public JsonSerde(@Nullable JavaType targetType) {
+		this(targetType, JacksonUtils.enhancedObjectMapper());
 	}
 
 	public JsonSerde(ObjectMapper objectMapper) {
-		this(null, objectMapper);
+		this((JavaType) null, objectMapper);
+	}
+
+	public JsonSerde(@Nullable TypeReference<? super T> targetType, ObjectMapper objectMapper) {
+		this(targetType == null ? null : objectMapper.constructType(targetType.getType()), objectMapper);
+	}
+
+	public JsonSerde(@Nullable Class<? super T> targetType, ObjectMapper objectMapper) {
+		this(targetType == null ? null : objectMapper.constructType(targetType), objectMapper);
 	}
 
-	@SuppressWarnings("unchecked")
-	public JsonSerde(@Nullable Class<? super T> targetTypeArg, @Nullable ObjectMapper objectMapperArg) {
-		ObjectMapper objectMapper = objectMapperArg;
-		Class<T> targetType = (Class<T>) targetTypeArg;
-		if (objectMapper == null) {
-			objectMapper = JacksonUtils.enhancedObjectMapper();
+	public JsonSerde(@Nullable JavaType targetTypeArg, @Nullable ObjectMapper objectMapperArg) {
+		ObjectMapper objectMapper = objectMapperArg == null ? JacksonUtils.enhancedObjectMapper() : objectMapperArg;
+		JavaType actualJavaType;
+		if (targetTypeArg != null) {
+			actualJavaType = targetTypeArg;
 		}
-		this.jsonSerializer = new JsonSerializer<>(objectMapper);
-		if (targetType == null) {
-			targetType = (Class<T>) ResolvableType.forClass(getClass()).getSuperType().resolveGeneric(0);
+		else {
+			Class<?> resolvedGeneric = ResolvableType.forClass(getClass()).getSuperType().resolveGeneric(0);
+			actualJavaType = resolvedGeneric != null ? objectMapper.constructType(resolvedGeneric) : null;
 		}
-		this.jsonDeserializer = new JsonDeserializer<>(targetType, objectMapper);
+		this.jsonSerializer = new JsonSerializer<>(actualJavaType, objectMapper);
+		this.jsonDeserializer = new JsonDeserializer<>(actualJavaType, objectMapper);
 	}
 
 	public JsonSerde(JsonSerializer<T> jsonSerializer, JsonDeserializer<T> jsonDeserializer) {
@@ -107,6 +125,42 @@ public Deserializer<T> deserializer() {
 		return this.jsonDeserializer;
 	}
 
+	/**
+	 * Copies this serde with same configuration, except new target type is used.
+	 * @param newTargetType type reference forced for serialization, and used as default for deserialization, not null
+	 * @param <X> new deserialization result type and serialization source type
+	 * @return new instance of serde with type changes
+	 * @since 2.6
+	 */
+	public <X> JsonSerde<X> copyWithType(Class<? super X> newTargetType) {
+		return new JsonSerde<>(this.jsonSerializer.copyWithType(newTargetType),
+			this.jsonDeserializer.copyWithType(newTargetType));
+	}
+
+	/**
+	 * Copies this serde with same configuration, except new target type reference is used.
+	 * @param newTargetType type reference forced for serialization, and used as default for deserialization, not null
+	 * @param <X> new deserialization result type and serialization source type
+	 * @return new instance of serde with type changes
+	 * @since 2.6
+	 */
+	public <X> JsonSerde<X> copyWithType(TypeReference<? super X> newTargetType) {
+		return new JsonSerde<>(this.jsonSerializer.copyWithType(newTargetType),
+			this.jsonDeserializer.copyWithType(newTargetType));
+	}
+
+	/**
+	 * Copies this serde with same configuration, except new target java type is used.
+	 * @param newTargetType java type forced for serialization, and used as default for deserialization, not null
+	 * @param <X> new deserialization result type and serialization source type
+	 * @return new instance of serde with type changes
+	 * @since 2.6
+	 */
+	public <X> JsonSerde<X> copyWithType(JavaType newTargetType) {
+		return new JsonSerde<>(this.jsonSerializer.copyWithType(newTargetType),
+			this.jsonDeserializer.copyWithType(newTargetType));
+	}
+
 	// Fluent API
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -33,7 +33,10 @@
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
 
 /**
  * Generic {@link org.apache.kafka.common.serialization.Serializer Serializer} for sending
@@ -63,17 +66,32 @@ public class JsonSerializer<T> implements Serializer<T> {
 
 	protected boolean addTypeInfo = true; // NOSONAR
 
+	private ObjectWriter writer;
+
 	protected Jackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper(); // NOSONAR
 
 	private boolean typeMapperExplicitlySet = false;
 
 	public JsonSerializer() {
-		this(JacksonUtils.enhancedObjectMapper());
+		this((JavaType) null, JacksonUtils.enhancedObjectMapper());
+	}
+
+	public JsonSerializer(TypeReference<? super T> targetType) {
+		this(targetType, JacksonUtils.enhancedObjectMapper());
 	}
 
 	public JsonSerializer(ObjectMapper objectMapper) {
+		this((JavaType) null, objectMapper);
+	}
+
+	public JsonSerializer(TypeReference<? super T> targetType, ObjectMapper objectMapper) {
+		this(targetType == null ? null : objectMapper.constructType(targetType.getType()), objectMapper);
+	}
+
+	public JsonSerializer(JavaType targetType, ObjectMapper objectMapper) {
 		Assert.notNull(objectMapper, "'objectMapper' must not be null.");
 		this.objectMapper = objectMapper;
+		this.writer = objectMapper.writerFor(targetType);
 	}
 
 	public boolean isAddTypeInfo() {
@@ -174,7 +192,7 @@ public byte[] serialize(String topic, @Nullable T data) {
 			return null;
 		}
 		try {
-			return this.objectMapper.writeValueAsBytes(data);
+			return this.writer.writeValueAsBytes(data);
 		}
 		catch (IOException ex) {
 			throw new SerializationException("Can't serialize data [" + data + "] for topic [" + topic + "]", ex);
@@ -186,6 +204,43 @@ public void close() {
 		// No-op
 	}
 
+	/**
+	 * Copies this serializer with same configuration, except new target type reference is used.
+	 * @param newTargetType type reference forced for serialization, not null
+	 * @param <X> new serialization source type
+	 * @return new instance of serializer with type changes
+	 * @since 2.6
+	 */
+	public <X> JsonSerializer<X> copyWithType(Class<? super X> newTargetType) {
+		return copyWithType(this.objectMapper.constructType(newTargetType));
+	}
+
+	/**
+	 * Copies this serializer with same configuration, except new target type reference is used.
+	 * @param newTargetType type reference forced for serialization, not null
+	 * @param <X> new serialization source type
+	 * @return new instance of serializer with type changes
+	 * @since 2.6
+	 */
+	public <X> JsonSerializer<X> copyWithType(TypeReference<? super X> newTargetType) {
+		return copyWithType(this.objectMapper.constructType(newTargetType.getType()));
+	}
+
+	/**
+	 * Copies this serializer with same configuration, except new target java type is used.
+	 * @param newTargetType java type forced for serialization, not null
+	 * @param <X> new serialization source type
+	 * @return new instance of serializer with type changes
+	 * @since 2.6
+	 */
+	public <X> JsonSerializer<X> copyWithType(JavaType newTargetType) {
+		JsonSerializer<X> result = new JsonSerializer<>(newTargetType, this.objectMapper);
+		result.addTypeInfo = this.addTypeInfo;
+		result.typeMapper = this.typeMapper;
+		result.typeMapperExplicitlySet = this.typeMapperExplicitlySet;
+		return result;
+	}
+
 	// Fluent API
 
 	/**