[GOBBLIN-1726] Avro 1.9 upgrade of Gobblin OSS (apache#3581)

* [Branch avro_1_9] Avro 1.9 upgrade compatible change - replaced deprecated public APIs with the compatible APIs (apache#3349)

* Avro 1.9 upgrade compatible change - replaced deprecated public APIs with the compatible APIs.

* Avro 1.9 upgrade compatible change - Replaced guava library import from avro shaded with direct guava libraries

* Applied Gobblin codestyle formatting.

Co-authored-by: Lei <[email protected]>

* Avro 1.9 upgrade compatible change - upgraded jackson mapper to 2.x and kept jackson mapper 1.x for modules with hive, helix library dependency. (apache#3368)

* Avro 1.9 upgrade compatible change - upgraded jackson mapper to 2.x and kept jackson mapper 1.x for modules with hive, helix library dependency.

* Changes for upgrade Avro 1.9.2 and leverges hive with avro changes from https://linkedin.jfrog.io/artifactory/gobblin-hive (apache#3458)

* Use helper-all v0.2.74 to solve issues around default values. (apache#3469)

The latest version of helper-all fixes the issues seen before w.r.t.
default values, so we can now revert the code and the *.avsc files back
to how they used to be, with two minor exceptions:

1. Check Schema equality using their .toString() representations. Doing
   it the old way works for two out of the three instances, but one of
   them fails, for reasons I haven't figured out yet.

2. Add a `"default":null` piece to recursive_schema_1_converted.avsc.
   This is harmless, and is caused by the fact that the compatibility
   helper always adds it if it's a valid default for the schema. See
   the comments for FieldBuilder19.setDefault():
   https://github.com/linkedin/avro-util/blob/b9e89c55980ea8e5fd3c8d8da362d7195dd2a99c/helper/impls/helper-impl-19/src/main/java/com/linkedin/avroutil1/compatibility/avro19/FieldBuilder19.java#L69

To verify that the files are otherwise the same as before:
```
$ for file in gobblin-core-base/src/test/resources/converter/*.avsc; do
> git show 928e018:$file > /tmp/before
> diff <(jq . </tmp/before) <(jq . <$file)
> done
```

* [Branch avro_1_9] Avro 1.9 upgrade compatible change - replaced deprecated public APIs with the compatible APIs (apache#3349)

* Avro 1.9 upgrade compatible change - replaced deprecated public APIs with the compatible APIs.

* Avro 1.9 upgrade compatible change - Replaced guava library import from avro shaded with direct guava libraries

* Applied Gobblin codestyle formatting.

Co-authored-by: Lei <[email protected]>

* Avro 1.9 upgrade compatible change - upgraded jackson mapper to 2.x and kept jackson mapper 1.x for modules with hive, helix library dependency. (apache#3368)

* Avro 1.9 upgrade compatible change - upgraded jackson mapper to 2.x and kept jackson mapper 1.x for modules with hive, helix library dependency.

* Changes for upgrade Avro 1.9.2 and leverges hive with avro changes from https://linkedin.jfrog.io/artifactory/gobblin-hive (apache#3458)

* Use helper-all v0.2.74 to solve issues around default values. (apache#3469)

The latest version of helper-all fixes the issues seen before w.r.t.
default values, so we can now revert the code and the *.avsc files back
to how they used to be, with two minor exceptions:

1. Check Schema equality using their .toString() representations. Doing
   it the old way works for two out of the three instances, but one of
   them fails, for reasons I haven't figured out yet.

2. Add a `"default":null` piece to recursive_schema_1_converted.avsc.
   This is harmless, and is caused by the fact that the compatibility
   helper always adds it if it's a valid default for the schema. See
   the comments for FieldBuilder19.setDefault():
   https://github.com/linkedin/avro-util/blob/b9e89c55980ea8e5fd3c8d8da362d7195dd2a99c/helper/impls/helper-impl-19/src/main/java/com/linkedin/avroutil1/compatibility/avro19/FieldBuilder19.java#L69

To verify that the files are otherwise the same as before:
```
$ for file in gobblin-core-base/src/test/resources/converter/*.avsc; do
> git show 928e018:$file > /tmp/before
> diff <(jq . </tmp/before) <(jq . <$file)
> done
```

* [Branch avro_1_9] Avro 1.9 upgrade compatible change - replaced deprecated public APIs with the compatible APIs (apache#3349)

* Avro 1.9 upgrade compatible change - replaced deprecated public APIs with the compatible APIs.

* Avro 1.9 upgrade compatible change - Replaced guava library import from avro shaded with direct guava libraries

* Applied Gobblin codestyle formatting.

Co-authored-by: Lei <[email protected]>

* Avro 1.9 upgrade compatible change - upgraded jackson mapper to 2.x and kept jackson mapper 1.x for modules with hive, helix library dependency. (apache#3368)

* Avro 1.9 upgrade compatible change - upgraded jackson mapper to 2.x and kept jackson mapper 1.x for modules with hive, helix library dependency.

* Changes for upgrade Avro 1.9.2 and leverges hive with avro changes from https://linkedin.jfrog.io/artifactory/gobblin-hive (apache#3458)

* Use helper-all v0.2.74 to solve issues around default values. (apache#3469)

The latest version of helper-all fixes the issues seen before w.r.t.
default values, so we can now revert the code and the *.avsc files back
to how they used to be, with two minor exceptions:

1. Check Schema equality using their .toString() representations. Doing
   it the old way works for two out of the three instances, but one of
   them fails, for reasons I haven't figured out yet.

2. Add a `"default":null` piece to recursive_schema_1_converted.avsc.
   This is harmless, and is caused by the fact that the compatibility
   helper always adds it if it's a valid default for the schema. See
   the comments for FieldBuilder19.setDefault():
   https://github.com/linkedin/avro-util/blob/b9e89c55980ea8e5fd3c8d8da362d7195dd2a99c/helper/impls/helper-impl-19/src/main/java/com/linkedin/avroutil1/compatibility/avro19/FieldBuilder19.java#L69

To verify that the files are otherwise the same as before:
```
$ for file in gobblin-core-base/src/test/resources/converter/*.avsc; do
> git show 928e018:$file > /tmp/before
> diff <(jq . </tmp/before) <(jq . <$file)
> done
```

* Merging apache/gobblin master with avro_1_9

* Use helper-all v0.2.74 to solve issues around default values. (apache#3469)

The latest version of helper-all fixes the issues seen before w.r.t.
default values, so we can now revert the code and the *.avsc files back
to how they used to be, with two minor exceptions:

1. Check Schema equality using their .toString() representations. Doing
   it the old way works for two out of the three instances, but one of
   them fails, for reasons I haven't figured out yet.

2. Add a `"default":null` piece to recursive_schema_1_converted.avsc.
   This is harmless, and is caused by the fact that the compatibility
   helper always adds it if it's a valid default for the schema. See
   the comments for FieldBuilder19.setDefault():
   https://github.com/linkedin/avro-util/blob/b9e89c55980ea8e5fd3c8d8da362d7195dd2a99c/helper/impls/helper-impl-19/src/main/java/com/linkedin/avroutil1/compatibility/avro19/FieldBuilder19.java#L69

To verify that the files are otherwise the same as before:
```
$ for file in gobblin-core-base/src/test/resources/converter/*.avsc; do
> git show 928e018:$file > /tmp/before
> diff <(jq . </tmp/before) <(jq . <$file)
> done
```

* Added deprecated json method using AvroCompatibilityHelper

* Removed unused import and replaced Integer.valueOf with Integer.parseInt

* Exclude com.linkedin.hive dependency from gradle build files similar to org.apache.hive

* Repalce direct avro field creation with AvroCompatibilityHelper.createSchemaField

* Removed extra dependency. Addressed review comment - removed jcenter() repository

* Upgrade AvroCompatHelper version

* Removed the code that are actually moved to AvroHiveTypeUtils.java in the master branch

* Addresssed review comments: replaced getObjectProps/getObjectProp with AvroCompatibilityHelper methods

* Fix for test failure

Co-authored-by: Lei <[email protected]>
Co-authored-by: Sreeram Ramachandran <[email protected]>

Author: 3 people

gobblin-cluster/build.gradle

@@ -50,6 +50,7 @@ dependencies {
   compile (externalDependency.helix) {
     exclude group: 'io.dropwizard.metrics', module: 'metrics-core'
   }
+  compile externalDependency.jacksonMapperAsl
 
   runtimeOnly project(":gobblin-modules:gobblin-service-kafka")
 

gobblin-compaction/build.gradle

@@ -46,6 +46,7 @@ dependencies {
 
   runtimeOnly(externalDependency.hiveService) {
     exclude group: 'org.apache.hive', module: 'hive-exec'
+    exclude group: 'com.linkedin.hive', module: 'hive-exec'
   }
   runtimeOnly externalDependency.hiveJdbc
   runtimeOnly externalDependency.hiveMetastore

gobblin-compaction/src/main/java/org/apache/gobblin/compaction/mapreduce/avro/FieldAttributeBasedDeltaFieldsProvider.java

@@ -17,6 +17,7 @@
 
 package org.apache.gobblin.compaction.mapreduce.avro;
 
+import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
@@ -27,6 +28,7 @@
 import org.apache.avro.generic.GenericRecord;
 import org.apache.hadoop.conf.Configuration;
 import org.codehaus.jackson.JsonFactory;
+import org.codehaus.jackson.JsonNode;
 import org.codehaus.jackson.JsonParser;
 import org.codehaus.jackson.map.ObjectMapper;
 import org.codehaus.jackson.node.ObjectNode;
@@ -79,7 +81,9 @@ public List<String> getDeltaFieldNames(GenericRecord record) {
   private List<String> getDeltaFieldNamesForNewSchema(Schema originalSchema) {
     List<String> deltaFields = new ArrayList<>();
     for (Field field : originalSchema.getFields()) {
-      String deltaAttributeField = field.getJsonProp(this.attributeField).getValueAsText();
+      // Avro 1.9 compatible change - replaced deprecated public api getJsonProp with AvroCompatibilityHelper methods
+      String deltaAttributeField = AvroCompatibilityHelper.getFieldPropAsJsonString(field, this.attributeField, 
+          true, false);
       ObjectNode objectNode = getDeltaPropValue(deltaAttributeField);
       if (objectNode == null || objectNode.get(this.deltaPropName) == null) {
         continue;
@@ -98,7 +102,8 @@ private ObjectNode getDeltaPropValue(String json) {
       JsonParser jp = jf.createJsonParser(json);
       ObjectMapper objMap = new ObjectMapper(jf);
       jp.setCodec(objMap);
-      return (ObjectNode) jp.readValueAsTree();
+      JsonNode jsonNode = jp.readValueAsTree();
+      return (ObjectNode) objMap.readTree(jsonNode.asText());
     } catch (IOException e) {
       return null;
     }

gobblin-compaction/src/main/java/org/apache/gobblin/compaction/mapreduce/avro/MRCompactorAvroKeyDedupJobRunner.java

@@ -37,6 +37,7 @@
 import org.apache.avro.mapred.AvroKey;
 import org.apache.avro.mapred.AvroValue;
 import org.apache.avro.mapreduce.AvroJob;
+import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper;
 import org.apache.commons.io.FilenameUtils;
 import org.apache.gobblin.compaction.dataset.Dataset;
 import org.apache.gobblin.compaction.mapreduce.MRCompactorJobRunner;
@@ -198,7 +199,8 @@ public static Optional<Schema> getKeySchemaFromRecord(Schema record) {
     for (Field field : record.getFields()) {
       Optional<Schema> newFieldSchema = getKeySchema(field);
       if (newFieldSchema.isPresent()) {
-        fields.add(new Field(field.name(), newFieldSchema.get(), field.doc(), field.defaultValue()));
+        fields.add(AvroCompatibilityHelper.createSchemaField(field.name(), newFieldSchema.get(), field.doc(),
+            AvroUtils.getCompatibleDefaultValue(field)));
       }
     }
     if (!fields.isEmpty()) {

gobblin-compaction/src/test/java/org/apache/gobblin/compaction/mapreduce/avro/FieldAttributeBasedDeltaFieldsProviderTest.java

@@ -25,7 +25,7 @@
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
-import avro.shaded.com.google.common.collect.Lists;
+import com.google.common.collect.Lists;
 
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;

gobblin-core-base/build.gradle

@@ -31,6 +31,7 @@ dependencies {
   compile externalDependency.avroMapredH2
   compile externalDependency.commonsCodec
   compile externalDependency.avro
+  compile externalDependency.avroCompatHelper
   compile externalDependency.guava
   compile externalDependency.slf4j
   compile externalDependency.typesafeConfig

gobblin-core-base/src/main/java/org/apache/gobblin/converter/filter/AvroSchemaFieldRemover.java

@@ -22,14 +22,15 @@
 
 import org.apache.avro.Schema;
 import org.apache.avro.Schema.Field;
-import org.codehaus.jackson.JsonNode;
+import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper;
 
 import com.google.common.base.Preconditions;
 import com.google.common.base.Splitter;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
-import static org.apache.gobblin.util.AvroUtils.convertFieldToSchemaWithProps;
+import org.apache.gobblin.util.AvroSchemaUtils;
+import org.apache.gobblin.util.AvroUtils;
 
 
 /**
@@ -108,7 +109,7 @@ private Schema removeFields(Schema schema, Map<String, Schema> schemaMap) {
   private Schema removeFieldsFromRecords(Schema schema, Map<String, Schema> schemaMap) {
 
     Schema newRecord = Schema.createRecord(schema.getName(), schema.getDoc(), schema.getNamespace(), schema.isError());
-    convertFieldToSchemaWithProps(schema.getJsonProps(), newRecord);
+    AvroSchemaUtils.copySchemaProperties(schema, newRecord);
 
     // Put an incomplete schema into schemaMap to avoid re-processing a recursive field.
     // The fields in the incomplete schema will be populated once the current schema is completely processed.
@@ -119,15 +120,16 @@ private Schema removeFieldsFromRecords(Schema schema, Map<String, Schema> schema
       if (!this.shouldRemove(field)) {
         Field newField;
         if (this.children.containsKey(field.name())) {
-          newField = new Field(field.name(), this.children.get(field.name()).removeFields(field.schema(), schemaMap),
-              field.doc(), field.defaultValue());
+          newField = AvroCompatibilityHelper.createSchemaField(field.name(),
+              this.children.get(field.name()).removeFields(field.schema(), schemaMap),
+              field.doc(), AvroUtils.getCompatibleDefaultValue(field));
         } else {
-          newField = new Field(field.name(), DO_NOTHING_INSTANCE.removeFields(field.schema(), schemaMap), field.doc(),
-              field.defaultValue());
-        }
-        for (Map.Entry<String, JsonNode> stringJsonNodeEntry : field.getJsonProps().entrySet()) {
-          newField.addProp(stringJsonNodeEntry.getKey(), stringJsonNodeEntry.getValue());
+          newField = AvroCompatibilityHelper.createSchemaField(field.name(),
+              DO_NOTHING_INSTANCE.removeFields(field.schema(), schemaMap), field.doc(),
+              AvroUtils.getCompatibleDefaultValue(field));
         }
+        // Avro 1.9 compatible change - replaced deprecated public api getJsonProps with AvroCompatibilityHelper methods
+        AvroSchemaUtils.copyFieldProperties(field, newField);
         newFields.add(newField);
       }
     }

gobblin-core-base/src/main/java/org/apache/gobblin/converter/filter/GobblinTrackingEventFlattenFilterConverter.java

@@ -26,6 +26,7 @@
 import org.apache.avro.Schema.Field;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericRecord;
+import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper;
 
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
@@ -85,10 +86,11 @@ public Converter init(WorkUnitState workUnitState) {
       String curFieldName = field.name();
       if (!field.schema().getType().equals(Schema.Type.MAP)) {
         if (fieldsRenameMap.containsKey(curFieldName)) {
-          newFields.add(
-              new Schema.Field(fieldsRenameMap.get(curFieldName), field.schema(), field.doc(), field.defaultValue()));
+          newFields.add(AvroCompatibilityHelper.createSchemaField(fieldsRenameMap.get(curFieldName), field.schema(),
+              field.doc(), AvroUtils.getCompatibleDefaultValue(field)));
         } else {
-          newFields.add(new Schema.Field(curFieldName, field.schema(), field.doc(), field.defaultValue()));
+          newFields.add(AvroCompatibilityHelper.createSchemaField(curFieldName, field.schema(), field.doc(),
+              AvroUtils.getCompatibleDefaultValue(field)));
         }
         this.nonMapFields.add(curFieldName);
       } else {
@@ -102,7 +104,7 @@ public Converter init(WorkUnitState workUnitState) {
     for (String fieldToFlatten : ConfigUtils.getStringList(config, FIELDS_TO_FLATTEN)) {
       String newFieldName =
           this.fieldsRenameMap.containsKey(fieldToFlatten) ? this.fieldsRenameMap.get(fieldToFlatten) : fieldToFlatten;
-      newFields.add(new Field(newFieldName, Schema.create(Schema.Type.STRING), "", null));
+      newFields.add(AvroCompatibilityHelper.createSchemaField(newFieldName, Schema.create(Schema.Type.STRING), "", null));
     }
 
     return this;

gobblin-core-base/src/main/java/org/apache/gobblin/test/SequentialTestSource.java

@@ -29,7 +29,7 @@
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
 
-import avro.shaded.com.google.common.base.Throwables;
+import com.google.common.base.Throwables;
 import lombok.extern.slf4j.Slf4j;
 
 import org.apache.gobblin.configuration.SourceState;

gobblin-core-base/src/test/java/org/apache/gobblin/converter/filter/AvroSchemaFieldRemoverTest.java

@@ -37,16 +37,16 @@ public class AvroSchemaFieldRemoverTest {
   public void testRemoveFields() throws IllegalArgumentException, IOException {
     Schema convertedSchema1 = convertSchema("/converter/recursive_schema_1.avsc", "YwchQiH.OjuzrLOtmqLW");
     Schema expectedSchema1 = parseSchema("/converter/recursive_schema_1_converted.avsc");
-    Assert.assertEquals(convertedSchema1, expectedSchema1);
+    Assert.assertEquals(convertedSchema1.toString(), expectedSchema1.toString());
 
     Schema convertedSchema2 =
         convertSchema("/converter/recursive_schema_2.avsc", "FBuKC.wIINqII.lvaerUEKxBQUWg,eFQjDj.TzuYZajb");
     Schema expectedSchema2 = parseSchema("/converter/recursive_schema_2_converted.avsc");
-    Assert.assertEquals(convertedSchema2, expectedSchema2);
+    Assert.assertEquals(convertedSchema2.toString(), expectedSchema2.toString());
 
     Schema convertedSchema3 = convertSchema("/converter/recursive_schema_2.avsc", "field.that.does.not.exist");
     Schema expectedSchema3 = parseSchema("/converter/recursive_schema_2_not_converted.avsc");
-    Assert.assertEquals(convertedSchema3, expectedSchema3);
+    Assert.assertEquals(convertedSchema3.toString(), expectedSchema3.toString());
   }
 
   private Schema parseSchema(String schemaFile) throws IOException {

gobblin-core-base/src/test/java/org/apache/gobblin/converter/filter/GobblinTrackingEventFlattenFilterConverterTest.java

@@ -45,12 +45,13 @@ public void testSchemaConversion()
     Schema output = converter.convertSchema(
         new Schema.Parser().parse(getClass().getClassLoader().getResourceAsStream("GobblinTrackingEvent.avsc")),
         workUnitState);
-    Assert.assertEquals(output, new Schema.Parser().parse(
+    Schema parsedSchema = new Schema.Parser().parse(
         "{\"type\":\"record\",\"name\":\"GobblinTrackingEvent\",\"namespace\":\"org.apache.gobblin.metrics\",\"fields\":"
             + "[{\"name\":\"timestamp\",\"type\":\"long\",\"doc\":\"Time at which event was created.\",\"default\":0},"
             + "{\"name\":\"namespace\",\"type\":[\"string\",\"null\"],\"doc\":\"Namespace used for filtering of events.\"},"
             + "{\"name\":\"name\",\"type\":\"string\",\"doc\":\"Event name.\"},{\"name\":\"field1\",\"type\":\"string\",\"doc\":\"\"},"
-            + "{\"name\":\"field2\",\"type\":\"string\",\"doc\":\"\"}]}"));
+            + "{\"name\":\"field2\",\"type\":\"string\",\"doc\":\"\"}]}");
+    Assert.assertEquals(output.toString(), parsedSchema.toString());
 
     props.put(GobblinTrackingEventFlattenFilterConverter.class.getSimpleName() + "."
         + GobblinTrackingEventFlattenFilterConverter.FIELDS_RENAME_MAP, "name:eventName,field1:field3");
@@ -61,11 +62,12 @@ public void testSchemaConversion()
     Schema output2 = converter.convertSchema(
         new Schema.Parser().parse(getClass().getClassLoader().getResourceAsStream("GobblinTrackingEvent.avsc")),
         workUnitState2);
-    Assert.assertEquals(output2, new Schema.Parser().parse(
+    parsedSchema = new Schema.Parser().parse(
         "{\"type\":\"record\",\"name\":\"GobblinTrackingEvent\",\"namespace\":\"org.apache.gobblin.metrics\",\"fields\":"
             + "[{\"name\":\"timestamp\",\"type\":\"long\",\"doc\":\"Time at which event was created.\",\"default\":0},"
             + "{\"name\":\"namespace\",\"type\":[\"string\",\"null\"],\"doc\":\"Namespace used for filtering of events.\"},"
             + "{\"name\":\"eventName\",\"type\":\"string\",\"doc\":\"Event name.\"},{\"name\":\"field3\",\"type\":\"string\",\"doc\":\"\"},"
-            + "{\"name\":\"field2\",\"type\":\"string\",\"doc\":\"\"}]}"));
+            + "{\"name\":\"field2\",\"type\":\"string\",\"doc\":\"\"}]}");
+    Assert.assertEquals(output2.toString(), parsedSchema.toString());
   }
 }