GH-680: Use enhanced ObjectMapper by default

spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java

@@ -94,7 +94,7 @@ public class DefaultKafkaHeaderMapper extends AbstractKafkaHeaderMapper {
 	 * @see #DefaultKafkaHeaderMapper(ObjectMapper)
 	 */
 	public DefaultKafkaHeaderMapper() {
-		this(new ObjectMapper());
+		this(JacksonUtils.enhancedObjectMapper());
 	}
 
 	/**

spring-kafka/src/main/java/org/springframework/kafka/support/JacksonUtils.java

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.support;
+
+import org.springframework.beans.BeanUtils;
+import org.springframework.util.ClassUtils;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.MapperFeature;
+import com.fasterxml.jackson.databind.Module;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * The utilities for Jackson {@link ObjectMapper} instances.
+ *
+ * @author Artem Bilan
+ *
+ * @since 2.3
+ */
+public final class JacksonUtils {
+
+	private static final String UNUSED = "unused";
+
+	/**
+	 * Factory for {@link ObjectMapper} instances with registered well-known modules
+	 * and disabled {@link MapperFeature#DEFAULT_VIEW_INCLUSION} and
+	 * {@link DeserializationFeature#FAIL_ON_UNKNOWN_PROPERTIES} features.
+	 * The {@link ClassUtils#getDefaultClassLoader()} is used for loading module classes.
+	 * @return the {@link ObjectMapper} instance.
+	 */
+	public static ObjectMapper enhancedObjectMapper() {
+		return enhancedObjectMapper(ClassUtils.getDefaultClassLoader());
+	}
+
+	/**
+	 * Factory for {@link ObjectMapper} instances with registered well-known modules
+	 * and disabled {@link MapperFeature#DEFAULT_VIEW_INCLUSION} and
+	 * {@link DeserializationFeature#FAIL_ON_UNKNOWN_PROPERTIES} features.
+	 * @param classLoader the {@link ClassLoader} for modules to register.
+	 * @return the {@link ObjectMapper} instance.
+	 */
+	public static ObjectMapper enhancedObjectMapper(ClassLoader classLoader) {
+		ObjectMapper objectMapper = new ObjectMapper();
+		objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
+		objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+		registerWellKnownModulesIfAvailable(objectMapper, classLoader);
+		return objectMapper;
+	}
+
+	@SuppressWarnings("unchecked")
+	private static void registerWellKnownModulesIfAvailable(ObjectMapper objectMapper, ClassLoader classLoader) {
+		try {
+			Class<? extends Module> jdk8Module = (Class<? extends Module>)
+					ClassUtils.forName("com.fasterxml.jackson.datatype.jdk8.Jdk8Module", classLoader);
+			objectMapper.registerModule(BeanUtils.instantiateClass(jdk8Module));
+		}
+		catch (@SuppressWarnings(UNUSED) ClassNotFoundException ex) {
+			// jackson-datatype-jdk8 not available
+		}
+
+		try {
+			Class<? extends Module> javaTimeModule = (Class<? extends Module>)
+					ClassUtils.forName("com.fasterxml.jackson.datatype.jsr310.JavaTimeModule", classLoader);
+			objectMapper.registerModule(BeanUtils.instantiateClass(javaTimeModule));
+		}
+		catch (@SuppressWarnings(UNUSED) ClassNotFoundException ex) {
+			// jackson-datatype-jsr310 not available
+		}
+
+		// Joda-Time present?
+		if (ClassUtils.isPresent("org.joda.time.LocalDate", classLoader)) {
+			try {
+				Class<? extends Module> jodaModule = (Class<? extends Module>)
+						ClassUtils.forName("com.fasterxml.jackson.datatype.joda.JodaModule", classLoader);
+				objectMapper.registerModule(BeanUtils.instantiateClass(jodaModule));
+			}
+			catch (@SuppressWarnings(UNUSED) ClassNotFoundException ex) {
+				// jackson-datatype-joda not available
+			}
+		}
+
+		// Kotlin present?
+		if (ClassUtils.isPresent("kotlin.Unit", classLoader)) {
+			try {
+				Class<? extends Module> kotlinModule = (Class<? extends Module>)
+						ClassUtils.forName("com.fasterxml.jackson.module.kotlin.KotlinModule", classLoader);
+				objectMapper.registerModule(BeanUtils.instantiateClass(kotlinModule));
+			}
+			catch (@SuppressWarnings(UNUSED) ClassNotFoundException ex) {
+				//jackson-module-kotlin not available
+			}
+		}
+	}
+
+	private JacksonUtils() {
+	}
+
+}

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

@@ -28,6 +28,7 @@
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.data.web.JsonProjectingMethodInterceptorFactory;
+import org.springframework.kafka.support.JacksonUtils;
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
@@ -40,6 +41,7 @@
  * {@link ProjectionFactory} to bind incoming messages to projection interfaces.
  *
  * @author Oliver Gierke
+ * @author Artem Bilan
  *
  * @since 2.1.1
  */
@@ -49,6 +51,15 @@ public class ProjectingMessageConverter extends MessagingMessageConverter {
 
 	private final MessagingMessageConverter delegate;
 
+	/**
+	 * Creates a new {@link ProjectingMessageConverter} using a
+	 * {@link JacksonUtils#enhancedObjectMapper()} by default.
+	 * @since 2.3
+	 */
+	public ProjectingMessageConverter() {
+		this(JacksonUtils.enhancedObjectMapper());
+	}
+
 	/**
 	 * Creates a new {@link ProjectingMessageConverter} using the given {@link ObjectMapper}.
 	 * @param mapper must not be {@literal null}.
@@ -100,11 +111,11 @@ private static byte[] getAsByteArray(Object source) {
 		Assert.notNull(source, "Source must not be null");
 
 		if (source instanceof String) {
-			return String.class.cast(source).getBytes(StandardCharsets.UTF_8);
+			return ((String) source).getBytes(StandardCharsets.UTF_8);
 		}
 
 		if (source instanceof byte[]) {
-			return byte[].class.cast(source);
+			return (byte[]) source;
 		}
 
 		throw new ConversionException(String.format("Unsupported payload type '%s'. Expected 'String' or 'byte[]'",

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

@@ -24,15 +24,14 @@
 import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.apache.kafka.common.utils.Bytes;
 
+import org.springframework.kafka.support.JacksonUtils;
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper.TypePrecedence;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JavaType;
-import com.fasterxml.jackson.databind.MapperFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.type.TypeFactory;
 
@@ -53,9 +52,7 @@ public class StringJsonMessageConverter extends MessagingMessageConverter {
 	private Jackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper();
 
 	public StringJsonMessageConverter() {
-		this(new ObjectMapper());
-		this.objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
-		this.objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+		this(JacksonUtils.enhancedObjectMapper());
 	}
 
 	public StringJsonMessageConverter(ObjectMapper objectMapper) {
@@ -112,8 +109,8 @@ protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type)
 		}
 
 		JavaType javaType = this.typeMapper.getTypePrecedence().equals(TypePrecedence.INFERRED)
-			? TypeFactory.defaultInstance().constructType(type)
-			: this.typeMapper.toJavaType(record.headers());
+				? TypeFactory.defaultInstance().constructType(type)
+				: this.typeMapper.toJavaType(record.headers());
 		if (javaType == null) { // no headers
 			javaType = TypeFactory.defaultInstance().constructType(type);
 		}

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

@@ -19,13 +19,13 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Map;
-import java.util.function.Consumer;
 
 import org.apache.kafka.common.errors.SerializationException;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.serialization.Deserializer;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.kafka.support.JacksonUtils;
 import org.springframework.kafka.support.converter.AbstractJavaTypeMapper;
 import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper;
 import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper;
@@ -35,9 +35,7 @@
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
 
-import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JavaType;
-import com.fasterxml.jackson.databind.MapperFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectReader;
 
@@ -152,10 +150,7 @@ public JsonDeserializer(Class<? super T> targetType) {
 	 * @since 2.2
 	 */
 	public JsonDeserializer(Class<? super T> targetType, boolean useHeadersIfPresent) {
-		this(targetType, new ObjectMapper(), useHeadersIfPresent, om -> {
-			om.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
-			om.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
-		});
+		this(targetType, JacksonUtils.enhancedObjectMapper(), useHeadersIfPresent);
 	}
 
 	/**
@@ -176,13 +171,9 @@ public JsonDeserializer(Class<? super T> targetType, ObjectMapper objectMapper)
 	 * type if not.
 	 * @since 2.2
 	 */
-	public JsonDeserializer(@Nullable Class<? super T> targetType, ObjectMapper objectMapper, boolean useHeadersIfPresent) {
-		this(targetType, objectMapper, useHeadersIfPresent, om -> { });
-	}
-
 	@SuppressWarnings("unchecked")
-	private JsonDeserializer(@Nullable Class<? super T> targetType, ObjectMapper objectMapper, boolean useHeadersIfPresent,
-			Consumer<ObjectMapper> configurer) {
+	public JsonDeserializer(@Nullable Class<? super T> targetType, ObjectMapper objectMapper,
+			boolean useHeadersIfPresent) {
 
 		Assert.notNull(objectMapper, "'objectMapper' must not be null.");
 		this.objectMapper = objectMapper;
@@ -193,7 +184,6 @@ private JsonDeserializer(@Nullable Class<? super T> targetType, ObjectMapper obj
 		Assert.isTrue(this.targetType != null || useHeadersIfPresent,
 				"'targetType' cannot be null if 'useHeadersIfPresent' is false");
 
-		configurer.accept(this.objectMapper);
 		if (this.targetType != null) {
 			this.reader = this.objectMapper.readerFor(this.targetType);
 		}

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

@@ -23,11 +23,10 @@
 import org.apache.kafka.common.serialization.Serializer;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.kafka.support.JacksonUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
-import com.fasterxml.jackson.databind.DeserializationFeature;
-import com.fasterxml.jackson.databind.MapperFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 /**
@@ -69,9 +68,7 @@ public JsonSerde(@Nullable Class<? super T> targetTypeArg, @Nullable ObjectMappe
 		ObjectMapper objectMapper = objectMapperArg;
 		Class<T> targetType = (Class<T>) targetTypeArg;
 		if (objectMapper == null) {
-			objectMapper = new ObjectMapper();
-			objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
-			objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+			objectMapper = JacksonUtils.enhancedObjectMapper();
 		}
 		this.jsonSerializer = new JsonSerializer<>(objectMapper);
 		if (targetType == null) {

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

@@ -24,6 +24,7 @@
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.serialization.Serializer;
 
+import org.springframework.kafka.support.JacksonUtils;
 import org.springframework.kafka.support.converter.AbstractJavaTypeMapper;
 import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper;
 import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper;
@@ -32,8 +33,6 @@
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
 
-import com.fasterxml.jackson.databind.DeserializationFeature;
-import com.fasterxml.jackson.databind.MapperFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 /**
@@ -69,9 +68,7 @@ public class JsonSerializer<T> implements Serializer<T> {
 	private boolean typeMapperExplicitlySet = false;
 
 	public JsonSerializer() {
-		this(new ObjectMapper());
-		this.objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
-		this.objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+		this(JacksonUtils.enhancedObjectMapper());
 	}
 
 	public JsonSerializer(ObjectMapper objectMapper) {

src/reference/asciidoc/kafka.adoc

@@ -2299,8 +2299,11 @@ JsonDeserializer<Thing> thingDeserializer = new JsonDeserializer<>(Thing.class);
 ====
 
 You can customize both `JsonSerializer` and `JsonDeserializer` with an `ObjectMapper`.
-You can also extend them to implement some particular configuration logic in the
-`configure(Map<String, ?> configs, boolean isKey)` method.
+You can also extend them to implement some particular configuration logic in the `configure(Map<String, ?> configs, boolean isKey)` method.
+
+Starting with version 2.3, all the JSON-aware components are configured by default with a `JacksonUtils.enhancedObjectMapper()` instance, which comes with the `MapperFeature.DEFAULT_VIEW_INCLUSION` and `DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES` features disabled.
+Also such an instance is supplied with well-known modules for custom data types, such a Java time and Kotlin support.
+See `JacksonUtils.enhancedObjectMapper()` JavaDocs for more information.
 
 Starting with version 2.1, you can convey type information in record `Headers`, allowing the handling of multiple types.
 In addition, you can configure the serializer and deserializer by using the following Kafka properties:

src/reference/asciidoc/whats-new.adoc

@@ -2,14 +2,14 @@
 
 This section covers the changes made from version 2.2 to version 2.3.
 
-[[kafka-client-2.1]]
+[[kafka-client-2.2]]
 ==== Kafka Client Version
 
 This version requires the 2.2.0 `kafka-clients` or higher.
 
 ==== Listener Container Changes
 
-Previously, error handlers received `ListenerExectionFailedException` (with the actual listener exception as the `cause`) when the listener was invoked using a listener adapter (such as `@KafkaListener` s).
+Previously, error handlers received `ListenerExecutionFailedException` (with the actual listener exception as the `cause`) when the listener was invoked using a listener adapter (such as `@KafkaListener` s).
 Exceptions thrown by native `GenericMessageListener` s were passed to the error handler unchanged.
 Now a `ListenerExecutionFailedException` is always the argument (with the actual listener exception as the `cause`), which provides access to the container's `group.id` property.
 
@@ -40,3 +40,8 @@ See <<configuring-topics>> for more information.
 
 You can now perform additional configuration of the `StreamsBuilderFactoryBean` created by `@EnableKafkaStreams`.
 See <<streams-config, Streams Configuration>> for more information.
+
+==== JSON Components Changes
+
+Now all the JSON-aware components are configured by default with a Jackson `ObjectMapper` produced by the `JacksonUtils.enhancedObjectMapper()`.
+See its JavaDocs and <<serdes>> for more information.