[GOBBLIN-2237]Make AvroUtils compatible with avro 1.10 (#4152)

* made avro get field compatible with avro 1.10

* addressed comments

* addressed comments

* made debug logs

Author: pratapaditya04

gobblin-utility/src/main/java/org/apache/gobblin/util/AvroUtils.java

@@ -306,9 +306,12 @@ private static void getFieldHelper(Map<String, Object> retVal,
       if (data instanceof Map) {
         val = getObjectFromMap((Map)data, pathList.get(field));
       } else if (data instanceof List) {
-        val = getObjectFromArray((List)data, Integer.parseInt(pathList.get(field)));
+        val = getObjectFromArray((List) data, Integer.parseInt(pathList.get(field)));
+      } else if (data instanceof GenericRecord) {
+        val = getSafeField((GenericRecord) data, pathList.get(field));
       } else {
-        val = ((GenericRecord)data).get(pathList.get(field));
+        log.warn("Unexpected data type '{}' encountered while traversing field '{}'. Value: {}", data.getClass().getName(),
+            field, data);
       }
 
       if (val != null) {
@@ -340,8 +343,20 @@ private static void getFieldHelper(Map<String, Object> retVal,
       return;
     }
 
-    AvroUtils.getFieldHelper(retVal, ((GenericRecord) data).get(pathList.get(field)), pathList, ++field);
-    return;
+    if (data instanceof GenericRecord) {
+      Object next = getSafeField((GenericRecord) data, pathList.get(field));
+      getFieldHelper(retVal, next, pathList, field + 1);
+    }
+  }
+
+  private static Object getSafeField(GenericRecord record, String fieldName) {
+    if (record == null || fieldName == null) return null;
+    try{
+      return record.get(fieldName);
+    } catch (AvroRuntimeException e) {
+      log.debug("Unable to get field {} from record {}", fieldName, record, e);
+      return null;
+    }
   }
 
   /**

gobblin-utility/src/test/java/org/apache/gobblin/util/AvroUtilsTest.java

@@ -247,7 +247,7 @@ public void testSwitchNamespace() {
         AvroCompatibilityHelper.createSchemaField("key", Schema.create(Schema.Type.LONG), "", 0L);
     field1.addProp("primaryKey", "true");
     fieldList.add(field1);
-    Schema.Field field2 = 
+    Schema.Field field2 =
         AvroCompatibilityHelper.createSchemaField("double", Schema.create(Schema.Type.DOUBLE), "", 0.0);
     fieldList.add(field2);
 
@@ -656,4 +656,38 @@ public void testDropRecursiveSchema()
 
     }
   }
+  @Test
+  public void testGetFieldValue_existingField() {
+    String schemaStr = "{\"type\":\"record\",\"name\":\"TestRecord\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"}]}";
+    Schema schema = new Schema.Parser().parse(schemaStr);
+    GenericRecord record = new GenericData.Record(schema);
+    record.put("id", "123");
+
+    Map<String, Object> result = AvroUtils.getMultiFieldValue(record, "id");
+    Assert.assertEquals("123", result.get("id"));
+  }
+
+  @Test
+  public void testGetFieldValue_missingField_returnsNull() {
+    String schemaStr = "{\"type\":\"record\",\"name\":\"TestRecord\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"}]}";
+    Schema schema = new Schema.Parser().parse(schemaStr);
+    GenericRecord record = new GenericData.Record(schema);
+    record.put("id", "123");
+
+    Map<String, Object> result = AvroUtils.getMultiFieldValue(record, "nonexistent");
+    Assert.assertTrue(result.isEmpty());
+  }
+
+  @Test
+  public void testNestedFieldAccess() {
+    String schemaStr = "{\"type\":\"record\",\"name\":\"Outer\",\"fields\":[{\"name\":\"inner\",\"type\":{\"type\":\"record\",\"name\":\"Inner\",\"fields\":[{\"name\":\"value\",\"type\":\"int\"}]}}]}";
+    Schema schema = new Schema.Parser().parse(schemaStr);
+    GenericRecord inner = new GenericData.Record(schema.getField("inner").schema());
+    inner.put("value", 42);
+    GenericRecord outer = new GenericData.Record(schema);
+    outer.put("inner", inner);
+
+    Map<String, Object> result = AvroUtils.getMultiFieldValue(outer, "inner.value");
+    Assert.assertEquals(42, result.get("inner.value"));
+  }
 }