SAMZA-2620: Jackson upgrade.

build.gradle

@@ -150,7 +150,7 @@ project(':samza-api') {
 
   dependencies {
     compile "org.apache.commons:commons-lang3:$commonsLang3Version"
-    compile "org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
     compile "com.google.guava:guava:$guavaVersion"
     compile "com.google.code.gson:gson:$gsonVersion"
     compile "org.slf4j:slf4j-api:$slf4jVersion"
@@ -190,7 +190,7 @@ project(":samza-core_$scalaSuffix") {
     compile "org.apache.commons:commons-collections4:$apacheCommonsCollections4Version"
     compile "org.apache.commons:commons-lang3:$commonsLang3Version"
     compile "commons-io:commons-io:$commonsIoVersion"
-    compile "org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
     compile "org.eclipse.jetty:jetty-webapp:$jettyVersion"
     compile "org.scala-lang:scala-library:$scalaVersion"
     compile "org.slf4j:slf4j-api:$slf4jVersion"
@@ -222,7 +222,7 @@ project(":samza-azure_$scalaSuffix") {
     compile "com.azure:azure-storage-blob:12.0.1"
     compile "com.microsoft.azure:azure-storage:5.3.1"
     compile "com.microsoft.azure:azure-eventhubs:1.0.1"
-    compile "com.fasterxml.jackson.core:jackson-core:2.10.0"
+    compile "com.fasterxml.jackson.core:jackson-core:$jacksonVersion"
     compile "io.dropwizard.metrics:metrics-core:3.1.2"
     compile "org.apache.avro:avro:$avroVersion"
     compile project(':samza-api')
@@ -244,8 +244,7 @@ project(":samza-aws_$scalaSuffix") {
     compile "com.amazonaws:amazon-kinesis-client:1.7.5"
     compile "com.amazonaws:amazon-kinesis-producer:0.10.0"
     compile "io.dropwizard.metrics:metrics-core:3.1.2"
-    compile "org.codehaus.jackson:jackson-core-asl:1.9.7"
-    compile "org.codehaus.jackson:jackson-mapper-asl:1.9.7"
+    compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
     compile project(':samza-api')
     compile project(":samza-core_$scalaSuffix")
     compile "org.slf4j:slf4j-api:$slf4jVersion"
@@ -393,7 +392,7 @@ project(":samza-kafka_$scalaSuffix") {
     compile "org.scala-lang:scala-library:$scalaVersion"
     compile "com.101tec:zkclient:$zkClientVersion"
     compile "org.apache.zookeeper:zookeeper:$zookeeperVersion"
-    compile "org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
     compile "org.apache.kafka:kafka_$scalaSuffix:$kafkaVersion"
     compile "org.apache.kafka:kafka-clients:$kafkaVersion"
     testCompile "org.apache.kafka:kafka_$scalaSuffix:$kafkaVersion:test"
@@ -405,6 +404,14 @@ project(":samza-kafka_$scalaSuffix") {
 
     // Logging in tests is good.
     testRuntime "org.slf4j:slf4j-simple:$slf4jVersion"
+
+    constraints {
+      compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
+      compile "com.fasterxml.jackson.core:jackson-annotations:$jacksonVersion"
+      compile "com.fasterxml.jackson.module:jackson-module-scala_$scalaSuffix:$jacksonVersion"
+      compile "com.fasterxml.jackson.dataformat:jackson-dataformat-csv:$jacksonVersion"
+      compile "com.fasterxml.jackson.datatype:jackson-datatype-jdk8:$jacksonVersion"
+    }
   }
 
   test {
@@ -428,7 +435,7 @@ project(":samza-log4j_$scalaSuffix") {
     compile "org.slf4j:slf4j-log4j12:$slf4jVersion"
     compile project(':samza-api')
     compile project(":samza-core_$scalaSuffix")
-    compile "org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
     testCompile "junit:junit:$junitVersion"
   }
 }
@@ -443,7 +450,7 @@ project(":samza-log4j2_$scalaSuffix") {
     compile "org.apache.logging.log4j:log4j-slf4j-impl:$log4j2Version"
     compile project(':samza-api')
     compile project(":samza-core_$scalaSuffix")
-    compile "org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
     testCompile "junit:junit:$junitVersion"
   }
 }
@@ -467,7 +474,7 @@ project(":samza-yarn_$scalaSuffix") {
     compile "org.scala-lang:scala-library:$scalaVersion"
     compile "org.scala-lang:scala-compiler:$scalaVersion"
     compile "com.google.guava:guava:$guavaVersion"
-    compile "org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
     compile "commons-httpclient:commons-httpclient:$commonsHttpClientVersion"
     compile "org.apache.httpcomponents:httpclient:$httpClientVersion"
     compile("org.apache.hadoop:hadoop-yarn-api:$yarnVersion") {

gradle/dependency-versions.gradle

@@ -32,7 +32,7 @@
   guavaVersion = "23.0"
   hamcrestVersion = "1.3"
   httpClientVersion = "4.4.1"
-  jacksonVersion = "1.9.13"
+  jacksonVersion = "2.12.2"
   jerseyVersion = "2.22.1"
   jettyVersion = "9.4.20.v20190813"
   jodaTimeVersion = "2.2"

samza-api/src/main/java/org/apache/samza/serializers/JsonSerdeV2.java

@@ -19,9 +19,9 @@
 
 package org.apache.samza.serializers;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.samza.SamzaException;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.codehaus.jackson.type.TypeReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

samza-api/src/main/java/org/apache/samza/sql/SamzaSqlRelRecord.java

@@ -26,7 +26,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import org.apache.samza.annotation.InterfaceStability;
-import org.codehaus.jackson.annotate.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonProperty;
 
 
 /**

samza-api/src/main/java/org/apache/samza/startpoint/Startpoint.java

@@ -22,7 +22,7 @@
 import com.google.common.base.Objects;
 import java.time.Instant;
 import org.apache.samza.annotation.InterfaceStability;
-import org.codehaus.jackson.annotate.JsonTypeInfo;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
 
 /**
  * Startpoint represents a position in a stream partition.

samza-api/src/main/java/org/apache/samza/system/EndOfStreamMessage.java

@@ -19,8 +19,8 @@
 
 package org.apache.samza.system;
 
-import org.codehaus.jackson.annotate.JsonCreator;
-import org.codehaus.jackson.annotate.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
 
 /**
  *  The EndOfStreamMessage is a control message that is sent out to next stage

samza-api/src/main/java/org/apache/samza/system/WatermarkMessage.java

@@ -19,8 +19,8 @@
 
 package org.apache.samza.system;
 
-import org.codehaus.jackson.annotate.JsonCreator;
-import org.codehaus.jackson.annotate.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
 
 /**
  *  The WatermarkMessage is a control message that is sent out to next stage

samza-aws/src/main/java/org/apache/samza/system/kinesis/consumer/KinesisSystemConsumerOffset.java

@@ -19,10 +19,10 @@
 
 package org.apache.samza.system.kinesis.consumer;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.samza.SamzaException;
 import org.apache.samza.serializers.JsonSerdeV2;
-import org.codehaus.jackson.annotate.JsonCreator;
-import org.codehaus.jackson.annotate.JsonProperty;
 
 
 /**

samza-core/src/main/java/org/apache/samza/clustermanager/container/placement/ContainerPlacementMessageObjectMapper.java

@@ -23,21 +23,21 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.ObjectCodec;
+import com.fasterxml.jackson.core.Version;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.jsontype.NamedType;
+import com.fasterxml.jackson.databind.module.SimpleModule;
 import org.apache.samza.container.placement.ContainerPlacementMessage;
 import org.apache.samza.container.placement.ContainerPlacementRequestMessage;
 import org.apache.samza.container.placement.ContainerPlacementResponseMessage;
-import org.codehaus.jackson.JsonGenerator;
-import org.codehaus.jackson.JsonNode;
-import org.codehaus.jackson.JsonParser;
-import org.codehaus.jackson.ObjectCodec;
-import org.codehaus.jackson.Version;
-import org.codehaus.jackson.map.DeserializationContext;
-import org.codehaus.jackson.map.JsonDeserializer;
-import org.codehaus.jackson.map.JsonSerializer;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.codehaus.jackson.map.SerializerProvider;
-import org.codehaus.jackson.map.jsontype.NamedType;
-import org.codehaus.jackson.map.module.SimpleModule;
 
 /**
  * Object mapper for serializing and deserializing Container placement messages
@@ -100,25 +100,25 @@ public ContainerPlacementMessage deserialize(JsonParser jsonParser, Deserializat
         throws IOException {
       ObjectCodec oc = jsonParser.getCodec();
       JsonNode node = oc.readTree(jsonParser);
-      String subType = node.get("subType").getTextValue();
-      String deploymentId = node.get("deploymentId").getTextValue();
-      String processorId = node.get("processorId").getTextValue();
-      String destinationHost = node.get("destinationHost").getTextValue();
-      long timestamp = node.get("timestamp").getLongValue();
+      String subType = node.get("subType").textValue();
+      String deploymentId = node.get("deploymentId").textValue();
+      String processorId = node.get("processorId").textValue();
+      String destinationHost = node.get("destinationHost").textValue();
+      long timestamp = node.get("timestamp").longValue();
       Duration requestExpiry =
-          node.get("requestExpiry") == null ? null : Duration.ofMillis(node.get("requestExpiry").getLongValue());
+          node.get("requestExpiry") == null ? null : Duration.ofMillis(node.get("requestExpiry").longValue());
       ContainerPlacementMessage.StatusCode statusCode = null;
-      UUID uuid = UUID.fromString(node.get("uuid").getTextValue());
+      UUID uuid = UUID.fromString(node.get("uuid").textValue());
       for (ContainerPlacementMessage.StatusCode code : ContainerPlacementMessage.StatusCode.values()) {
-        if (code.name().equals(node.get("statusCode").getTextValue())) {
+        if (code.name().equals(node.get("statusCode").textValue())) {
           statusCode = code;
         }
       }
       ContainerPlacementMessage message = null;
       if (subType.equals(ContainerPlacementRequestMessage.class.getSimpleName())) {
         message = new ContainerPlacementRequestMessage(uuid, deploymentId, processorId, destinationHost, requestExpiry, timestamp);
       } else if (subType.equals(ContainerPlacementResponseMessage.class.getSimpleName())) {
-        String responseMessage = node.get("responseMessage").getTextValue();
+        String responseMessage = node.get("responseMessage").textValue();
         message = new ContainerPlacementResponseMessage(uuid, deploymentId, processorId, destinationHost, requestExpiry,
             statusCode, responseMessage, timestamp);
       }

samza-core/src/main/java/org/apache/samza/clustermanager/container/placement/ContainerPlacementMetadataStore.java

@@ -18,6 +18,7 @@
  */
 package org.apache.samza.clustermanager.container.placement;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import java.io.IOException;
@@ -36,7 +37,6 @@
 import org.apache.samza.container.placement.ContainerPlacementResponseMessage;
 import org.apache.samza.coordinator.metadatastore.NamespaceAwareCoordinatorStreamStore;
 import org.apache.samza.metadatastore.MetadataStore;
-import org.codehaus.jackson.map.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

samza-core/src/main/java/org/apache/samza/container/ContainerHeartbeatClient.java

@@ -19,6 +19,7 @@
 
 package org.apache.samza.container;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
@@ -27,7 +28,6 @@
 import java.util.stream.Collectors;
 import org.apache.samza.coordinator.CoordinationConstants;
 import org.apache.samza.util.HttpUtil;
-import org.codehaus.jackson.map.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

samza-core/src/main/java/org/apache/samza/container/ContainerHeartbeatResponse.java

@@ -19,8 +19,7 @@
 
 package org.apache.samza.container;
 
-import org.codehaus.jackson.annotate.JsonProperty;
-
+import com.fasterxml.jackson.annotation.JsonProperty;
 
 /**
  * Used to represent the heartbeat response between

samza-core/src/main/java/org/apache/samza/container/grouper/task/TaskPartitionAssignmentManager.java

@@ -26,8 +26,8 @@
 import org.apache.samza.serializers.Serde;
 import org.apache.samza.serializers.model.SamzaObjectMapper;
 import org.apache.samza.system.SystemStreamPartition;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.codehaus.jackson.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.core.type.TypeReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import java.io.IOException;

samza-core/src/main/java/org/apache/samza/coordinator/JobCoordinatorMetadataManager.java

@@ -18,6 +18,7 @@
  */
 package org.apache.samza.coordinator;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableSortedSet;
@@ -40,7 +41,6 @@
 import org.apache.samza.metrics.MetricsRegistry;
 import org.apache.samza.serializers.Serde;
 import org.apache.samza.serializers.model.SamzaObjectMapper;
-import org.codehaus.jackson.map.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java

@@ -18,6 +18,8 @@
  */
 package org.apache.samza.coordinator.metadatastore;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Sets;
@@ -49,8 +51,6 @@
 import org.apache.samza.system.SystemStreamPartition;
 import org.apache.samza.system.SystemStreamPartitionIterator;
 import org.apache.samza.util.CoordinatorStreamUtil;
-import org.codehaus.jackson.annotate.JsonProperty;
-import org.codehaus.jackson.map.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

samza-core/src/main/java/org/apache/samza/coordinator/server/LocalityServlet.java

@@ -18,6 +18,7 @@
  */
 package org.apache.samza.coordinator.server;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.IOException;
 import java.util.Optional;
 import javax.servlet.http.HttpServlet;
@@ -27,7 +28,6 @@
 import org.apache.samza.job.model.ProcessorLocality;
 import org.apache.samza.job.model.LocalityModel;
 import org.apache.samza.serializers.model.SamzaObjectMapper;
-import org.codehaus.jackson.map.ObjectMapper;
 
 
 /**

samza-core/src/main/java/org/apache/samza/execution/ExecutionPlan.java

@@ -55,7 +55,7 @@ public interface ExecutionPlan {
    * Returns the JSON representation of the plan.
    * @return JSON string
    * @throws Exception exception during JSON serialization, including {@link java.io.IOException}
-   *                   and {@link org.codehaus.jackson.JsonGenerationException}
+   *                   and {@link com.fasterxml.jackson.core.JsonGenerationException}
    */
   String getPlanAsJson() throws Exception;
 }

samza-core/src/main/java/org/apache/samza/execution/JobGraphJsonGenerator.java

@@ -19,6 +19,8 @@
 
 package org.apache.samza.execution;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import java.io.ByteArrayOutputStream;
 import java.util.ArrayList;
@@ -40,8 +42,6 @@
 import org.apache.samza.operators.spec.StreamTableJoinOperatorSpec;
 import org.apache.samza.table.descriptors.BaseTableDescriptor;
 import org.apache.samza.table.descriptors.TableDescriptor;
-import org.codehaus.jackson.annotate.JsonProperty;
-import org.codehaus.jackson.map.ObjectMapper;
 
 /**
  * This class generates the JSON representation of the {@link JobGraph}.

samza-core/src/main/java/org/apache/samza/job/model/JobModelUtil.java

@@ -30,7 +30,7 @@
 import org.apache.samza.serializers.model.SamzaObjectMapper;
 import org.apache.samza.system.SystemStream;
 import org.apache.samza.system.SystemStreamPartition;
-import org.codehaus.jackson.map.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
 /**
  * Utility class for the {@link JobModel}

samza-core/src/main/java/org/apache/samza/serializers/model/JsonContainerModelMixIn.java

@@ -19,16 +19,16 @@
 
 package org.apache.samza.serializers.model;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
 import java.util.Map;
 import org.apache.samza.container.TaskName;
 import org.apache.samza.job.model.TaskModel;
-import org.codehaus.jackson.annotate.JsonProperty;
 
 /**
  * A mix-in Jackson class to convert {@link org.apache.samza.job.model.ContainerModel} to JSON.
  * Notes:
  * 1) Constructor is not needed because this mixin is not used for deserialization. See {@link SamzaObjectMapper}.
- * 2) It is unnecessary to use {@link org.codehaus.jackson.annotate.JsonIgnoreProperties#ignoreUnknown()} here since
+ * 2) It is unnecessary to use {@link com.fasterxml.jackson.annotation.JsonIgnoreProperties#ignoreUnknown()} here since
  * {@link SamzaObjectMapper} already uses custom deserialization code for the
  * {@link org.apache.samza.job.model.ContainerModel}.
  * 3) See {@link SamzaObjectMapper} for more context about why the JSON keys are named in this specified way.