GH-3930: Add Jackson 3 support; deprecate Jackson 2

Fixes: #3930

* Manage  dependencies, similar way we do for Jackson 2
* Add Jackson 3 counterparts for existing Jackson 2 based classes (mostly copy/paste)
* Deprectate Jackson 2 classes
* Update tests
* Initial round of updates in docs

Signed-off-by: Soby Chacko <[email protected]>

Author: sobychacko

build.gradle

@@ -56,6 +56,7 @@ ext {
 	hamcrestVersion = '3.0'
 	hibernateValidationVersion = '8.0.2.Final'
 	jacksonBomVersion = '2.19.1'
+	jackson3Version = '3.0.0-rc5'
 	jaywayJsonPathVersion = '2.9.0'
 	junit4Version = '4.13.2'
 	junitJupiterVersion = '5.13.3'
@@ -110,6 +111,7 @@ allprojects {
 
 		imports {
 			mavenBom "com.fasterxml.jackson:jackson-bom:$jacksonBomVersion"
+			mavenBom "tools.jackson:jackson-bom:$jackson3Version"
 			mavenBom "org.junit:junit-bom:$junitJupiterVersion"
 			mavenBom "io.micrometer:micrometer-bom:$micrometerVersion"
 			mavenBom "io.micrometer:micrometer-tracing-bom:$micrometerTracingVersion"
@@ -263,6 +265,13 @@ project ('spring-kafka') {
 			exclude group: 'org.jetbrains.kotlin'
 		}
 
+		optionalApi 'tools.jackson.core:jackson-databind'
+		optionalApi 'tools.jackson.datatype:jackson-datatype-joda'
+		optionalApi 'tools.jackson.dataformat:jackson-dataformat-xml'
+		optionalApi('tools.jackson.module:jackson-module-kotlin') {
+			exclude group: 'org.jetbrains.kotlin'
+		}
+
 		// Spring Data projection message binding support
 		optionalApi ('org.springframework.data:spring-data-commons') {
 			exclude group: 'org.springframework'

samples/sample-01/README.adoc

@@ -2,7 +2,7 @@
 
 This sample demonstrates a simple producer and consumer; the producer sends objects of type `Foo1` and the consumer receives objects of type `Foo2` (the objects have the same field, `foo`).
 
-The producer uses a `JsonSerializer`; the consumer uses the `ByteArrayDeserializer`, together with a `JsonMessageConverter` which converts to the type of the listener method argument.
+The producer uses a `JacksonJsonSerializer`; the consumer uses the `ByteArrayDeserializer`, together with a `JacksonJsonMessageConverter` which converts to the type of the listener method argument.
 
 Run the application and use curl to send some data:
 
@@ -31,4 +31,4 @@ Console:
 ...
 2018-11-05 10:12:33.537  INFO 41635 --- [ fooGroup-0-C-1] com.example.Application                  : Received: Foo2 [foo=fail]
 2018-11-05 10:12:43.359  INFO 41635 --- [ dltGroup-0-C-1] com.example.Application                  : Received from DLT: {"foo":"fail"}
-----
+----

samples/sample-02/README.adoc

@@ -2,7 +2,7 @@
 
 This sample demonstrates a simple producer and a multi-method consumer; the producer sends objects of types `Foo1` and `Bar1` and the consumer receives objects of type `Foo2` and `Bar2` (the objects have the same field, `foo`).
 
-The producer uses a `JsonSerializer`; the consumer uses a `ByteArrayDeserializer`, together with a `ByteArrayJsonMessageConverter` which converts to the required type of the listener method argument.
+The producer uses a `JacksonJsonSerializer`; the consumer uses a `ByteArrayJacksonDeserializer`, together with a `ByteArrayJacksonJsonMessageConverter` which converts to the required type of the listener method argument.
 We can't infer the type in this case (because the type is used to choose the method to call).
 We therefore configure type mapping on the producer and consumer side.
 See the `application.yml` for the producer side and the `converter` bean on the consumer side.

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/class-level-kafkalistener.adoc

@@ -35,7 +35,7 @@ static class MultiListenerBean {
 Starting with version 2.1.3, you can designate a `@KafkaHandler` method as the default method that is invoked if there is no match on other methods.
 At most, one method can be so designated.
 When using `@KafkaHandler` methods, the payload must have already been converted to the domain object (so the match can be performed).
-Use a custom deserializer, the `JsonDeserializer`, or the `JsonMessageConverter` with its `TypePrecedence` set to `TYPE_ID`.
+Use a custom deserializer, the `JacksonJsonDeserializer`, or the `JacksonJsonMessageConverter` with its `TypePrecedence` set to `TYPE_ID`.
 See xref:kafka/serdes.adoc[Serialization, Deserialization, and Message Conversion] for more information.
 
 IMPORTANT: Due to some limitations in the way Spring resolves method arguments, a default `@KafkaHandler` cannot receive discrete headers; it must use the `ConsumerRecordMetadata` as discussed in xref:kafka/receiving-messages/listener-annotation.adoc#consumer-record-metadata[Consumer Record Metadata].

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/serdes.adoc

@@ -75,18 +75,18 @@ A `ToFromStringSerde` is also provided, for use with Kafka Streams.
 [[json-serde]]
 == JSON
 
-Spring for Apache Kafka also provides `JsonSerializer` and `JsonDeserializer` implementations that are based on the
+Spring for Apache Kafka also provides `JacksonJsonSerializer` and `JacksonJsonDeserializer` implementations that are based on the
 Jackson JSON object mapper.
-The `JsonSerializer` allows writing any Java object as a JSON `byte[]`.
-The `JsonDeserializer` requires an additional `Class<?> targetType` argument to allow the deserialization of a consumed `byte[]` to the proper target object.
-The following example shows how to create a `JsonDeserializer`:
+The `JacksonJsonSerializer` allows writing any Java object as a JSON `byte[]`.
+The `JacksonJsonDeserializer` requires an additional `Class<?> targetType` argument to allow the deserialization of a consumed `byte[]` to the proper target object.
+The following example shows how to create a `JacksonJsonDeserializer`:
 
 [source, java]
 ----
-JsonDeserializer<Thing> thingDeserializer = new JsonDeserializer<>(Thing.class);
+JacksonJsonDeserializer<Thing> thingDeserializer = new JacksonJsonDeserializer<>(Thing.class);
 ----
 
-You can customize both `JsonSerializer` and `JsonDeserializer` with an `ObjectMapper`.
+You can customize both `JacksonJsonSerializer` and `JacksonJsonDeserializer` with an `ObjectMapper`.
 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.
@@ -104,22 +104,22 @@ They have no effect if you have provided `Serializer` and `Deserializer` instanc
 [[serdes-json-config]]
 === Configuration Properties
 
-* `JsonSerializer.ADD_TYPE_INFO_HEADERS` (default `true`): You can set it to `false` to disable this feature on the `JsonSerializer` (sets the `addTypeInfo` property).
-* `JsonSerializer.TYPE_MAPPINGS` (default `empty`): See xref:kafka/serdes.adoc#serdes-mapping-types[Mapping Types].
-* `JsonDeserializer.USE_TYPE_INFO_HEADERS` (default `true`): You can set it to `false` to ignore headers set by the serializer.
-* `JsonDeserializer.REMOVE_TYPE_INFO_HEADERS` (default `true`): You can set it 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.
+* `JacksonJsonSerializer.ADD_TYPE_INFO_HEADERS` (default `true`): You can set it to `false` to disable this feature on the `JacksonJsonSerializer` (sets the `addTypeInfo` property).
+* `JacksonJsonSerializer.TYPE_MAPPINGS` (default `empty`): See xref:kafka/serdes.adoc#serdes-mapping-types[Mapping Types].
+* `JacksonJsonDeserializer.USE_TYPE_INFO_HEADERS` (default `true`): You can set it to `false` to ignore headers set by the serializer.
+* `JacksonJsonDeserializer.REMOVE_TYPE_INFO_HEADERS` (default `true`): You can set it to `false` to retain headers set by the serializer.
+* `JacksonJsonDeserializer.KEY_DEFAULT_TYPE`: Fallback type for deserialization of keys if no header information is present.
+* `JacksonJsonDeserializer.VALUE_DEFAULT_TYPE`: Fallback type for deserialization of values if no header information is present.
+* `JacksonJsonDeserializer.TRUSTED_PACKAGES` (default `java.util`, `java.lang`): Comma-delimited list of package patterns allowed for deserialization.
 `*` means deserializing all.
-* `JsonDeserializer.TYPE_MAPPINGS` (default `empty`): See xref:kafka/serdes.adoc#serdes-mapping-types[Mapping Types].
-* `JsonDeserializer.KEY_TYPE_METHOD` (default `empty`): See xref:kafka/serdes.adoc#serdes-type-methods[Using Methods to Determine Types].
-* `JsonDeserializer.VALUE_TYPE_METHOD` (default `empty`): See xref:kafka/serdes.adoc#serdes-type-methods[Using Methods to Determine Types].
+* `JacksonJsonDeserializer.TYPE_MAPPINGS` (default `empty`): See xref:kafka/serdes.adoc#serdes-mapping-types[Mapping Types].
+* `JacksonJsonDeserializer.KEY_TYPE_METHOD` (default `empty`): See xref:kafka/serdes.adoc#serdes-type-methods[Using Methods to Determine Types].
+* `JacksonJsonDeserializer.VALUE_TYPE_METHOD` (default `empty`): See xref:kafka/serdes.adoc#serdes-type-methods[Using Methods to Determine Types].
 
 Starting with version 2.2, the type information headers (if added by the serializer) are 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 earlier.
 
-See also xref:tips.adoc#tip-json[Customizing the JsonSerializer and JsonDeserializer].
+See also xref:tips.adoc#tip-json[Customizing the JacksonJsonSerializer and JacksonJsonDeserializer].
 
 IMPORTANT: Starting with version 2.8, if you construct the serializer or deserializer programmatically as shown in xref:kafka/serdes.adoc#prog-json[Programmatic Construction], the above properties will be applied by the factories, as long as you have not set any properties explicitly (using `set*()` methods or using the fluent API).
 Previously, when creating programmatically, the configuration properties were never applied; this is still the case if you explicitly set properties on the object directly.
@@ -409,7 +409,7 @@ Refer to the https://github.com/spring-projects/spring-retry[spring-retry] proje
 == Spring Messaging Message Conversion
 
 Although the `Serializer` and `Deserializer` API is quite simple and flexible from the low-level Kafka `Consumer` and `Producer` perspective, you might need more flexibility at the Spring Messaging level, when using either `@KafkaListener` or {spring-integration-url}/kafka.html[Spring Integration's Apache Kafka Support].
-To let you easily convert to and from `org.springframework.messaging.Message`, Spring for Apache Kafka provides a `MessageConverter` abstraction with the `MessagingMessageConverter` implementation and its `JsonMessageConverter` (and subclasses) customization.
+To let you easily convert to and from `org.springframework.messaging.Message`, Spring for Apache Kafka provides a `MessageConverter` abstraction with the `MessagingMessageConverter` implementation and its `JacksonJsonMessageConverter` (and subclasses) customization.
 You can inject the `MessageConverter` into a `KafkaTemplate` instance directly and by using `AbstractKafkaListenerContainerFactory` bean definition for the `@KafkaListener.containerFactory()` property.
 The following example shows how to do so:
 
@@ -443,15 +443,15 @@ With a class-level `@KafkaListener`, the payload type is used to select which `@
 
 [NOTE]
 ====
-On the consumer side, you can configure a `JsonMessageConverter`; it can handle `ConsumerRecord` values of type `byte[]`, `Bytes` and `String` so should be used in conjunction with a `ByteArrayDeserializer`, `BytesDeserializer` or `StringDeserializer`.
+On the consumer side, you can configure a `JacksonJsonMessageConverter`; it can handle `ConsumerRecord` values of type `byte[]`, `Bytes` and `String` so should be used in conjunction with a `ByteArrayDeserializer`, `BytesDeserializer` or `StringDeserializer`.
 (`byte[]` and `Bytes` are more efficient because they avoid an unnecessary `byte[]` to `String` conversion).
-You can also configure the specific subclass of `JsonMessageConverter` corresponding to the deserializer, if you so wish.
+You can also configure the specific subclass of `JacksonJsonMessageConverter` corresponding to the deserializer, if you so wish.
 
 On the producer side, when you use Spring Integration or the `KafkaTemplate.send(Message<?> message)` method (see xref:kafka/sending-messages.adoc#kafka-template[Using `KafkaTemplate`]), you must configure a message converter that is compatible with the configured Kafka `Serializer`.
 
-* `StringJsonMessageConverter` with `StringSerializer`
-* `BytesJsonMessageConverter` with `BytesSerializer`
-* `ByteArrayJsonMessageConverter` with `ByteArraySerializer`
+* `StringJacksonJsonMessageConverter` with `StringSerializer`
+* `BytesJacksonJsonMessageConverter` with `BytesSerializer`
+* `ByteArrayJacksonJsonMessageConverter` with `ByteArraySerializer`
 
 Again, using `byte[]` or `Bytes` is more efficient because they avoid a `String` to `byte[]` conversion.
 
@@ -513,7 +513,7 @@ public void projection(SomeSample in) {
 Accessor methods will be used to lookup the property name as field in the received JSON document by default.
 The `@JsonPath` expression allows customization of the value lookup, and even to define multiple JSON Path expressions, to look up values from multiple places until an expression returns an actual value.
 
-To enable this feature, use a `ProjectingMessageConverter` configured with an appropriate delegate converter (used for outbound conversion and converting non-projection interfaces).
+To enable this feature, use a `JacksonProjectingMessageConverter` configured with an appropriate delegate converter (used for outbound conversion and converting non-projection interfaces).
 You must also add `spring-data:spring-data-commons` and `com.jayway.jsonpath:json-path` to the classpath.
 
 When used as the parameter to a `@KafkaListener` method, the interface type is automatically passed to the converter as normal.
@@ -673,11 +673,11 @@ When using Spring Boot, this property name is `spring.kafka.consumer.properties.
 [[payload-conversion-with-batch]]
 == Payload Conversion with Batch Listeners
 
-You can also use a `JsonMessageConverter` within a `BatchMessagingMessageConverter` to convert batch messages when you use a batch listener container factory.
+You can also use a `JacksonJsonMessageConverter` within a `BatchMessagingMessageConverter` to convert batch messages when you use a batch listener container factory.
 See xref:kafka/serdes.adoc[Serialization, Deserialization, and Message Conversion] and xref:kafka/serdes.adoc#messaging-message-conversion[Spring Messaging Message Conversion] for more information.
 
 By default, the type for the conversion is inferred from the listener argument.
-If you configure the `JsonMessageConverter` with a `DefaultJackson2TypeMapper` that has its `TypePrecedence` set to `TYPE_ID` (instead of the default `INFERRED`), the converter uses the type information in headers (if present) instead.
+If you configure the `JacksonJsonMessageConverter` with a `DefaultJackson2TypeMapper` that has its `TypePrecedence` set to `TYPE_ID` (instead of the default `INFERRED`), the converter uses the type information in headers (if present) instead.
 This allows, for example, listener methods to be declared with interfaces instead of concrete classes.
 Also, the type converter supports mapping, so the deserialization can be to a different type than the source (as long as the data is compatible).
 This is also useful when you use xref:kafka/receiving-messages/class-level-kafkalistener.adoc[class-level `@KafkaListener` instances] where the payload must have already been converted to determine which method to invoke.

spring-kafka-docs/src/main/antora/modules/ROOT/pages/streams.adoc

@@ -151,24 +151,24 @@ streamsBuilderFactoryBean.addListener(new KafkaStreamsMicrometerListener(meterRe
 [[serde]]
 == Streams JSON Serialization and Deserialization
 
-For serializing and deserializing data when reading or writing to topics or state stores in JSON format, Spring for Apache Kafka provides a `JsonSerde` implementation that uses JSON, delegating to the `JsonSerializer` and `JsonDeserializer` described in xref:kafka/serdes.adoc[Serialization, Deserialization, and Message Conversion].
-The `JsonSerde` implementation provides the same configuration options through its constructor (target type or `ObjectMapper`).
-In the following example, we use the `JsonSerde` to serialize and deserialize the `Cat` payload of a Kafka stream (the `JsonSerde` can be used in a similar fashion wherever an instance is required):
+For serializing and deserializing data when reading or writing to topics or state stores in JSON format, Spring for Apache Kafka provides a `JacksonJsonSerde` implementation that uses JSON, delegating to the `JacksonJsonSerializer` and `JacksonJsonDeserializer` described in xref:kafka/serdes.adoc[Serialization, Deserialization, and Message Conversion].
+The `JacksonJsonSerde` implementation provides the same configuration options through its constructor (target type or `ObjectMapper`).
+In the following example, we use the `JacksonJsonSerde` to serialize and deserialize the `Cat` payload of a Kafka stream (the `JacksonJsonSerde` can be used in a similar fashion wherever an instance is required):
 
 [source,java]
 ----
-stream.through(Serdes.Integer(), new JsonSerde<>(Cat.class), "cats");
+stream.through(Serdes.Integer(), new JacksonJsonSerde<>(Cat.class), "cats");
 ----
 
 When constructing the serializer/deserializer programmatically for use in the producer/consumer factory, since version 2.3, you can use the fluent API, which simplifies configuration.
 
 [source, java]
 ----
 stream.through(
-    new JsonSerde<>(MyKeyType.class)
+    new JackonJsonSerde<>(MyKeyType.class)
         .forKeys()
         .noTypeInfo(),
-    new JsonSerde<>(MyValueType.class)
+    new JacksonJsonSerde<>(MyValueType.class)
         .noTypeInfo(),
     "myTypes");
 ----

spring-kafka-docs/src/main/java/org/springframework/kafka/jdocs/requestreply/Application.java

@@ -38,7 +38,7 @@
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
 import org.springframework.kafka.requestreply.RequestReplyTypedMessageFuture;
-import org.springframework.kafka.support.converter.ByteArrayJsonMessageConverter;
+import org.springframework.kafka.support.converter.ByteArrayJacksonJsonMessageConverter;
 import org.springframework.messaging.handler.annotation.SendTo;
 import org.springframework.messaging.support.MessageBuilder;
 
@@ -89,7 +89,7 @@ ReplyingKafkaTemplate<String, String, String> template(
         replyContainer.getContainerProperties().setGroupId("request.replies");
         ReplyingKafkaTemplate<String, String, String> template =
                 new ReplyingKafkaTemplate<>(pf, replyContainer);
-        template.setMessageConverter(new ByteArrayJsonMessageConverter());
+        template.setMessageConverter(new ByteArrayJacksonJsonMessageConverter());
         template.setDefaultTopic("requests");
         return template;
     }

spring-kafka-docs/src/main/kotlin/org/springframework/kafka/kdocs/requestreply/Application.kt

@@ -31,7 +31,7 @@ import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.kafka.core.ProducerFactory
 import org.springframework.kafka.requestreply.ReplyingKafkaTemplate
 import org.springframework.kafka.requestreply.RequestReplyTypedMessageFuture
-import org.springframework.kafka.support.converter.ByteArrayJsonMessageConverter
+import org.springframework.kafka.support.converter.ByteArrayJacksonJsonMessageConverter
 import org.springframework.messaging.handler.annotation.SendTo
 import org.springframework.messaging.support.MessageBuilder
 import java.util.concurrent.TimeUnit
@@ -76,7 +76,7 @@ class Application {
         val replyContainer = factory.createContainer("replies")
         replyContainer.containerProperties.setGroupId("request.replies")
         val template = ReplyingKafkaTemplate<String, String, String>(pf, replyContainer)
-        template.messageConverter = ByteArrayJsonMessageConverter()
+        template.messageConverter = ByteArrayJacksonJsonMessageConverter()
         template.setDefaultTopic("requests")
         return template
     }