Upgrade avro and move jackson from codehaus to fasterxml

build.gradle

@@ -142,6 +142,9 @@ project(':datastream-common') {
     compile "com.intellij:annotations:$intellijAnnotationsVersion"
     compile "com.google.guava:guava:$guavaVersion"
     compile "com.linkedin.kafka:kafka-clients:$kafkaVersion"
+    compile "com.fasterxml.jackson.core:jackson-annotations:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-core:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
     testCompile "org.mockito:mockito-core:$mockitoVersion"
   }
 }
@@ -316,8 +319,9 @@ project(':datastream-server') {
 
   dependencies {
     compile "com.101tec:zkclient:$zkclientVersion"
-    compile "org.codehaus.jackson:jackson-core-asl:$jacksonVersion"
-    compile "org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-annotations:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-core:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
 
     compile project(':datastream-server-api')
     compile project(':datastream-common')
@@ -352,8 +356,9 @@ project(':datastream-server-restli') {
     compile "com.linkedin.pegasus:r2-jetty:$pegasusVersion"
     compile "com.linkedin.parseq:parseq:$parseqVersion"
     compile "com.101tec:zkclient:$zkclientVersion"
-    compile "org.codehaus.jackson:jackson-core-asl:$jacksonVersion"
-    compile "org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-annotations:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-core:$jacksonVersion"
+    compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
     compile "org.apache.commons:commons-lang3:$commonslang3Version"
 
     compile project(':datastream-server')

datastream-common/src/main/java/com/linkedin/datastream/avrogenerator/AvroJson.java

@@ -13,9 +13,10 @@
 import java.util.Map;
 
 import org.apache.avro.Schema;
-import org.codehaus.jackson.JsonFactory;
-import org.codehaus.jackson.JsonGenerator;
-import org.codehaus.jackson.map.ObjectMapper;
+
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
 import com.google.common.base.CaseFormat;
 

datastream-common/src/main/java/com/linkedin/datastream/common/DatastreamEvent.java

@@ -6,12 +6,12 @@
 // CHECKSTYLE:OFF
 /**
  * Autogenerated by Avro
- * 
+ *
  * DO NOT EDIT DIRECTLY
  */
 package com.linkedin.datastream.common;
 
-@SuppressWarnings("all")
+@SuppressWarnings("serial")
 public class DatastreamEvent extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
   public static final org.apache.avro.Schema SCHEMA$ = org.apache.avro.Schema.parse("{\"type\":\"record\",\"name\":\"DatastreamEvent\",\"namespace\":\"com.linkedin.datastream.common\",\"fields\":[{\"name\":\"metadata\",\"type\":{\"type\":\"map\",\"values\":\"string\"},\"doc\":\"Event metadata.\"},{\"name\":\"key\",\"type\":\"bytes\",\"doc\":\"serialized key.\"},{\"name\":\"payload\",\"type\":\"bytes\",\"doc\":\"serialized payload data.\"},{\"name\":\"previous_payload\",\"type\":\"bytes\",\"doc\":\"serialized previous payload data.\"}]}");
   /** Event metadata. */
@@ -23,7 +23,7 @@ public class DatastreamEvent extends org.apache.avro.specific.SpecificRecordBase
   /** serialized previous payload data. */
   public java.nio.ByteBuffer previous_payload;
   public org.apache.avro.Schema getSchema() { return SCHEMA$; }
-  // Used by DatumWriter.  Applications should not call. 
+  // Used by DatumWriter.  Applications should not call.
   public java.lang.Object get(int field$) {
     switch (field$) {
     case 0: return metadata;
@@ -33,7 +33,7 @@ public java.lang.Object get(int field$) {
     default: throw new org.apache.avro.AvroRuntimeException("Bad index");
     }
   }
-  // Used by DatumReader.  Applications should not call. 
+  // Used by DatumReader.  Applications should not call.
   @SuppressWarnings(value="unchecked")
   public void put(int field$, java.lang.Object value$) {
     switch (field$) {

datastream-common/src/main/java/com/linkedin/datastream/common/JsonUtils.java

@@ -5,24 +5,26 @@
  */
 package com.linkedin.datastream.common;
 
+
 import java.io.IOException;
 import java.io.StringWriter;
 import java.time.Instant;
 
 import org.apache.commons.lang.Validate;
-import org.codehaus.jackson.JsonGenerator;
-import org.codehaus.jackson.JsonParser;
-import org.codehaus.jackson.annotate.JsonIgnore;
-import org.codehaus.jackson.map.DeserializationConfig;
-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.type.TypeReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+
 
 /**
  * Utility class for converting objects and JSON strings.
@@ -34,12 +36,11 @@ public final class JsonUtils {
 
   private static final ObjectMapper MAPPER = new ObjectMapper();
   static {
-    MAPPER.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+    MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
 
-    final DeserializationConfig config = MAPPER.getDeserializationConfig();
-    config.addMixInAnnotations(Datastream.class, IgnoreDatastreamSetPausedMixIn.class);
-    config.addMixInAnnotations(DatastreamSource.class, IgnoreDatastreamSourceSetPartitionsMixIn.class);
-    config.addMixInAnnotations(DatastreamDestination.class, IgnoreDatastreamDestinationSetPartitionsMixIn.class);
+    MAPPER.addMixIn(Datastream.class, IgnoreDatastreamSetPausedMixIn.class);
+    MAPPER.addMixIn(DatastreamSource.class, IgnoreDatastreamSourceSetPartitionsMixIn.class);
+    MAPPER.addMixIn(DatastreamDestination.class, IgnoreDatastreamDestinationSetPartitionsMixIn.class);
   }
 
   /**

datastream-file-connector/src/main/java/com/linkedin/datastream/connectors/file/diag/FilePositionKey.java

@@ -8,11 +8,12 @@
 import java.time.Instant;
 import java.util.Objects;
 
-import org.codehaus.jackson.annotate.JsonProperty;
-import org.codehaus.jackson.map.annotate.JsonDeserialize;
-import org.codehaus.jackson.map.annotate.JsonSerialize;
 import org.jetbrains.annotations.NotNull;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
 import com.linkedin.datastream.common.JsonUtils.InstantDeserializer;
 import com.linkedin.datastream.common.JsonUtils.InstantSerializer;
 import com.linkedin.datastream.common.diag.PositionKey;

datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/AbstractKafkaBasedConnectorTask.java

@@ -37,9 +37,9 @@
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
-import org.codehaus.jackson.type.TypeReference;
 import org.slf4j.Logger;
 
+import com.fasterxml.jackson.core.type.TypeReference;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Sets;
 

datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaConnector.java

@@ -18,10 +18,11 @@
 import org.apache.commons.lang.StringUtils;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.PartitionInfo;
-import org.codehaus.jackson.type.TypeReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+
 import com.linkedin.datastream.common.Datastream;
 import com.linkedin.datastream.common.DatastreamMetadataConstants;
 import com.linkedin.datastream.common.DatastreamSource;

datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaConnectorDiagUtils.java

@@ -12,9 +12,10 @@
 import java.util.Map;
 
 import org.apache.commons.lang3.StringUtils;
-import org.codehaus.jackson.type.TypeReference;
 import org.slf4j.Logger;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+
 import com.linkedin.datastream.common.JsonUtils;
 
 

datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaConsumerOffsetsResponse.java

@@ -8,7 +8,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import org.codehaus.jackson.annotate.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonProperty;
 
 
 /**

datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaDatastreamStatesResponse.java

@@ -11,17 +11,19 @@
 import java.util.Set;
 
 import org.apache.kafka.common.TopicPartition;
-import org.codehaus.jackson.JsonParser;
-import org.codehaus.jackson.Version;
-import org.codehaus.jackson.annotate.JsonPropertyOrder;
-import org.codehaus.jackson.map.DeserializationConfig;
-import org.codehaus.jackson.map.DeserializationContext;
-import org.codehaus.jackson.map.JsonDeserializer;
-import org.codehaus.jackson.map.KeyDeserializer;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.codehaus.jackson.map.annotate.JsonSerialize;
-import org.codehaus.jackson.map.module.SimpleModule;
-import org.codehaus.jackson.map.ser.ToStringSerializer;
+
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.Version;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.KeyDeserializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.databind.ser.std.ToStringSerializer;
 
 import com.linkedin.datastream.common.JsonUtils;
 import com.linkedin.datastream.server.FlushlessEventProducerHandler;
@@ -106,7 +108,7 @@ public static KafkaDatastreamStatesResponse fromJson(String json) {
     simpleModule.addDeserializer(TopicPartition.class, TopicPartitionDeserializer.getInstance());
     ObjectMapper mapper = new ObjectMapper();
     mapper.registerModule(simpleModule);
-    mapper.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+    mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
     return JsonUtils.fromJson(json, KafkaDatastreamStatesResponse.class, mapper);
   }
 
@@ -195,7 +197,7 @@ static TopicPartitionDeserializer getInstance() {
     @Override
     public TopicPartition deserialize(JsonParser jp, DeserializationContext ctxt)
         throws IOException {
-      String topicPartition = jp.getCodec().readTree(jp).getTextValue();
+      String topicPartition = ((JsonNode) jp.getCodec().readTree(jp)).asText();
       return topicPartitionFromString(topicPartition);
     }
   }

datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/KafkaTopicPartitionStatsResponse.java

@@ -10,8 +10,8 @@
 import java.util.Map;
 import java.util.Set;
 
-import org.codehaus.jackson.annotate.JsonCreator;
-import org.codehaus.jackson.annotate.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
 
 
 /**

datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/PausedSourcePartitionMetadata.java

@@ -11,7 +11,8 @@
 import java.util.function.BooleanSupplier;
 
 import org.apache.commons.lang.exception.ExceptionUtils;
-import org.codehaus.jackson.annotate.JsonPropertyOrder;
+
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
 
 
 /**

datastream-kafka-connector/src/test/java/com/linkedin/datastream/connectors/kafka/mirrormaker/TestKafkaConsumerOffsets.java

@@ -13,12 +13,13 @@
 import java.util.Optional;
 import java.util.stream.IntStream;
 
-import org.codehaus.jackson.type.TypeReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+
 import com.linkedin.datastream.common.Datastream;
 import com.linkedin.datastream.common.JsonUtils;
 import com.linkedin.datastream.common.PollUtils;

datastream-kafka-connector/src/test/java/com/linkedin/datastream/connectors/kafka/mirrormaker/TestKafkaMirrorMakerConnector.java

@@ -21,12 +21,12 @@
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.TopicPartition;
-import org.codehaus.jackson.type.TypeReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
+import com.fasterxml.jackson.core.type.TypeReference;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;

datastream-kafka-connector/src/test/java/com/linkedin/datastream/connectors/kafka/mirrormaker/TestKafkaTopicPartitionStats.java

@@ -15,10 +15,10 @@
 import java.util.Optional;
 import java.util.Set;
 
-import org.codehaus.jackson.type.TypeReference;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
+import com.fasterxml.jackson.core.type.TypeReference;
 import com.google.common.collect.ImmutableSet;
 
 import com.linkedin.datastream.common.Datastream;

datastream-server/src/main/java/com/linkedin/datastream/server/DatastreamTaskImpl.java

@@ -19,11 +19,12 @@
 import java.util.stream.Collectors;
 
 import org.apache.commons.lang.Validate;
-import org.codehaus.jackson.annotate.JsonIgnore;
 import org.jetbrains.annotations.TestOnly;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
+
 import com.linkedin.datastream.common.Datastream;
 import com.linkedin.datastream.common.DatastreamDestination;
 import com.linkedin.datastream.common.DatastreamMetadataConstants;

datastream-server/src/main/java/com/linkedin/datastream/server/providers/FileBasedPartitionThroughputProvider.java

@@ -14,9 +14,10 @@
 
 import org.apache.commons.lang.NotImplementedException;
 import org.apache.commons.lang3.StringUtils;
-import org.codehaus.jackson.JsonNode;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.codehaus.jackson.type.TypeReference;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
 import com.linkedin.datastream.server.ClusterThroughputInfo;
 import com.linkedin.datastream.server.DatastreamGroup;
@@ -87,7 +88,7 @@ private HashMap<String, ClusterThroughputInfo> readThroughputInfoFromFile(File f
     try {
       JsonNode root = mapper.readTree(file);
       JsonNode allStats = root.get(ROOT_NODE_NAME);
-      Iterator<String> clusterNames = allStats.getFieldNames();
+      Iterator<String> clusterNames = allStats.fieldNames();
 
       while (clusterNames.hasNext()) {
         String key = clusterNames.next();
@@ -127,7 +128,7 @@ private ClusterThroughputInfo getClusterThroughputInfoFromNode(ObjectMapper mapp
     HashMap<String, PartitionThroughputInfo> partitionInfoMap = new HashMap<>();
 
     try {
-      HashMap<String, String> partitionStats = mapper.readValue(clusterStats, mapTypeRef);
+      HashMap<String, String> partitionStats = mapper.readValue(clusterStats.toString(), mapTypeRef);
       for (String partition : partitionStats.keySet()) {
         String value = partitionStats.get(partition);
         String[] tokens = StringUtils.split(value, ",");

datastream-server/src/main/java/com/linkedin/datastream/server/providers/ZookeeperCheckpointProvider.java

@@ -16,10 +16,11 @@
 
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.Validate;
-import org.codehaus.jackson.type.TypeReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+
 import com.linkedin.datastream.common.JsonUtils;
 import com.linkedin.datastream.metrics.BrooklinHistogramInfo;
 import com.linkedin.datastream.metrics.BrooklinMeterInfo;

datastream-tools/src/main/java/com/linkedin/datastream/tools/DatastreamRestClientCli.java

@@ -18,8 +18,9 @@
 import org.apache.commons.cli.HelpFormatter;
 import org.apache.commons.cli.Options;
 import org.apache.commons.lang.StringUtils;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.codehaus.jackson.type.TypeReference;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
 import com.linkedin.data.template.StringMap;
 import com.linkedin.datastream.DatastreamRestClient;
@@ -63,7 +64,7 @@ private static void printDatastreams(boolean noformat, List<Datastream> streams)
         String jsonValue = DatastreamUtils.toJSON(s);
         if (!noformat) {
           Object json = mapper.readValue(jsonValue, Object.class);
-          jsonValue = mapper.defaultPrettyPrintingWriter().writeValueAsString(json);
+          jsonValue = mapper.writerWithDefaultPrettyPrinter().writeValueAsString(json);
         }
 
         System.out.println(jsonValue);

datastream-utils/src/main/java/com/linkedin/datastream/common/DatastreamUtils.java

@@ -21,7 +21,8 @@
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.Validate;
-import org.codehaus.jackson.type.TypeReference;
+
+import com.fasterxml.jackson.core.type.TypeReference;
 
 import com.linkedin.data.template.GetMode;
 import com.linkedin.data.template.StringMap;