public static Object getNestedFieldVal(GenericRecord record, String fieldName, boolean returnNullIfNotFound, boolean consistentLogicalTimestampEnabled) {
    String[] parts = fieldName.split("\\.");
    GenericRecord valueNode = record;

    for (int i = 0; i < parts.length; i++) {
      String part = parts[i];
      Object val;
      try {
        val = HoodieAvroUtils.getFieldVal(valueNode, part, returnNullIfNotFound);
      } catch (AvroRuntimeException e) {
        if (returnNullIfNotFound) {
          return null;
        } else {
          throw new HoodieException(
              fieldName + "(Part -" + parts[i] + ") field not found in record. Acceptable fields were :"
                  + valueNode.getSchema().getFields().stream().map(Field::name).collect(Collectors.toList()));
        }
      }

      if (i == parts.length - 1) {
        // return, if last part of name
        if (val == null) {
          return null;
        } else {
          Schema fieldSchema = valueNode.getSchema().getField(part).schema();
          return convertValueForSpecificDataTypes(fieldSchema, val, consistentLogicalTimestampEnabled);
        }
      } else {
        if (!(val instanceof GenericRecord)) {
          if (returnNullIfNotFound) {
            return null;
          } else {
            throw new HoodieException("Cannot find a record at part value :" + part);
          }
        } else {
          valueNode = (GenericRecord) val;
        }
      }
    }

    // This can only be reached if the length of parts is 0
    if (returnNullIfNotFound) {
      return null;
    } else {
      throw new HoodieException(
          fieldName + " field not found in record. Acceptable fields were :"
              + valueNode.getSchema().getFields().stream().map(Field::name).collect(Collectors.toList()));
    }
  }
@Test
  public void testGetNestedFieldValWithNestedField() {
    Schema nestedSchema = new Schema.Parser().parse(SCHEMA_WITH_NESTED_FIELD);
    GenericRecord rec = new GenericData.Record(nestedSchema);

    // test get .
    assertEquals(". field not found in record. Acceptable fields were :[firstname, lastname, student]",
        assertThrows(HoodieException.class, () ->
            HoodieAvroUtils.getNestedFieldVal(rec, ".", false, false)).getMessage());

    // test get fake_key
    assertEquals("fake_key(Part -fake_key) field not found in record. Acceptable fields were :[firstname, lastname, student]",
        assertThrows(HoodieException.class, () ->
            HoodieAvroUtils.getNestedFieldVal(rec, "fake_key", false, false)).getMessage());

    // test get student(null)
    assertNull(HoodieAvroUtils.getNestedFieldVal(rec, "student", false, false));

    // test get student
    GenericRecord studentRecord = new GenericData.Record(rec.getSchema().getField("student").schema());
    studentRecord.put("firstname", "person");
    rec.put("student", studentRecord);
    assertEquals(studentRecord, HoodieAvroUtils.getNestedFieldVal(rec, "student", false, false));

    // test get student.fake_key
    assertEquals("student.fake_key(Part -fake_key) field not found in record. Acceptable fields were :[firstname, lastname]",
        assertThrows(HoodieException.class, () ->
            HoodieAvroUtils.getNestedFieldVal(rec, "student.fake_key", false, false)).getMessage());

    // test get student.firstname
    assertEquals("person", HoodieAvroUtils.getNestedFieldVal(rec, "student.firstname", false, false));

    // test get student.lastname(null)
    assertNull(HoodieAvroUtils.getNestedFieldVal(rec, "student.lastname", false, false));

    // test get student.firstname.fake_key
    assertEquals("Cannot find a record at part value :firstname",
        assertThrows(HoodieException.class, () ->
            HoodieAvroUtils.getNestedFieldVal(rec, "student.firstname.fake_key", false, false)).getMessage());

    // test get student.lastname(null).fake_key
    assertEquals("Cannot find a record at part value :lastname",
        assertThrows(HoodieException.class, () ->
            HoodieAvroUtils.getNestedFieldVal(rec, "student.lastname.fake_key", false, false)).getMessage());
  }