KAFKA-15996: Improve JsonConverter performance (apache#14992)

Improve JsonConverter performance by using afterBurnModule of Jackson library. Reviewers: Divij Vaidya <[email protected]>, Mickael Maison <[email protected]>
patelsmit32123 · Dec 24, 2023 · 314de9f · 314de9f
1 parent ee65ca2
commit 314de9f
Show file tree

Hide file tree

Showing 6 changed files with 472 additions and 6 deletions.
diff --git a/build.gradle b/build.gradle
@@ -2706,6 +2706,7 @@ project(':jmh-benchmarks') {
     implementation project(':core')
     implementation project(':connect:api')
     implementation project(':connect:transforms')
+    implementation project(':connect:json')
     implementation project(':clients').sourceSets.test.output
     implementation project(':core').sourceSets.test.output
     implementation project(':server-common').sourceSets.test.output
@@ -2852,6 +2853,7 @@ project(':connect:json') {
 
     api libs.jacksonDatabind
     api libs.jacksonJDK8Datatypes
+    api libs.jacksonAfterburner
 
     implementation libs.slf4jApi
 

diff --git a/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java b/connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java
@@ -235,9 +235,21 @@ public Object toConnect(final Schema schema, final JsonNode value) {
     private final JsonDeserializer deserializer;
 
     public JsonConverter() {
+        this(true);
+    }
+
+    /**
+     * Creates a JsonConvert initializing serializer and deserializer.
+     *
+     * @param enableModules permits to enable/disable the registration of additional Jackson modules.
+     * <p>
+     * NOTE: This is visible only for testing
+     */
+    public JsonConverter(boolean enableModules) {
         serializer = new JsonSerializer(
             mkSet(),
-            JSON_NODE_FACTORY
+            JSON_NODE_FACTORY,
+            enableModules
         );
 
         deserializer = new JsonDeserializer(
@@ -246,7 +258,8 @@ public JsonConverter() {
                 // floating point numbers that cannot fit into float64
                 DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS
             ),
-            JSON_NODE_FACTORY
+            JSON_NODE_FACTORY,
+            enableModules
         );
     }
 

diff --git a/connect/json/src/main/java/org/apache/kafka/connect/json/JsonDeserializer.java b/connect/json/src/main/java/org/apache/kafka/connect/json/JsonDeserializer.java
@@ -38,7 +38,7 @@ public class JsonDeserializer implements Deserializer<JsonNode> {
      * Default constructor needed by Kafka
      */
     public JsonDeserializer() {
-        this(Collections.emptySet(), new JsonNodeFactory(true));
+        this(Collections.emptySet(), new JsonNodeFactory(true), true);
     }
 
     /**
@@ -50,11 +50,15 @@ public JsonDeserializer() {
      */
     JsonDeserializer(
         final Set<DeserializationFeature> deserializationFeatures,
-        final JsonNodeFactory jsonNodeFactory
+        final JsonNodeFactory jsonNodeFactory,
+        final boolean enableModules
     ) {
         objectMapper.enable(JsonReadFeature.ALLOW_LEADING_ZEROS_FOR_NUMBERS.mappedFeature());
         deserializationFeatures.forEach(objectMapper::enable);
         objectMapper.setNodeFactory(jsonNodeFactory);
+        if (enableModules) {
+            objectMapper.findAndRegisterModules();
+        }
     }
 
     @Override

diff --git a/connect/json/src/main/java/org/apache/kafka/connect/json/JsonSerializer.java b/connect/json/src/main/java/org/apache/kafka/connect/json/JsonSerializer.java
@@ -37,7 +37,7 @@ public class JsonSerializer implements Serializer<JsonNode> {
      * Default constructor needed by Kafka
      */
     public JsonSerializer() {
-        this(Collections.emptySet(), new JsonNodeFactory(true));
+        this(Collections.emptySet(), new JsonNodeFactory(true), true);
     }
 
     /**
@@ -49,10 +49,14 @@ public JsonSerializer() {
      */
     JsonSerializer(
         final Set<SerializationFeature> serializationFeatures,
-        final JsonNodeFactory jsonNodeFactory
+        final JsonNodeFactory jsonNodeFactory,
+        final boolean enableModules
     ) {
         serializationFeatures.forEach(objectMapper::enable);
         objectMapper.setNodeFactory(jsonNodeFactory);
+        if (enableModules) {
+            objectMapper.findAndRegisterModules();
+        }
     }
 
     @Override

diff --git a/gradle/dependencies.gradle b/gradle/dependencies.gradle
@@ -189,6 +189,7 @@ libs += [
   jacksonDataformatCsv: "com.fasterxml.jackson.dataformat:jackson-dataformat-csv:$versions.jackson",
   jacksonModuleScala: "com.fasterxml.jackson.module:jackson-module-scala_$versions.baseScala:$versions.jackson",
   jacksonJDK8Datatypes: "com.fasterxml.jackson.datatype:jackson-datatype-jdk8:$versions.jackson",
+  jacksonAfterburner: "com.fasterxml.jackson.module:jackson-module-afterburner:$versions.jackson",
   jacksonJaxrsJsonProvider: "com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:$versions.jackson",
   jaxAnnotationApi: "javax.annotation:javax.annotation-api:$versions.jaxAnnotation",
   jaxbApi: "javax.xml.bind:jaxb-api:$versions.jaxb",