From 47c5e518a756df815287502a50da9d73d28fc662 Mon Sep 17 00:00:00 2001
From: wangxianghu <wxhjsxz@126.com>
Date: Wed, 6 Jan 2021 19:49:17 +0800
Subject: [PATCH] [HUDI-1506] Fix wrong exception thrown in HoodieAvroUtils
 (#2405)

---
 .../org/apache/hudi/avro/HoodieAvroUtils.java |  4 ++-
 .../apache/hudi/avro/TestHoodieAvroUtils.java | 30 +++++++++++++++++++
 2 files changed, 33 insertions(+), 1 deletion(-)

diff --git a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java
index b1dcff90b..37a49eacc 100644
--- a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java
@@ -428,10 +428,12 @@ public class HoodieAvroUtils {
 
     if (returnNullIfNotFound) {
       return null;
-    } else {
+    } else if (valueNode.getSchema().getField(parts[i]) == null) {
       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()));
+    } else {
+      throw new HoodieException("The value of " + parts[i] + " can not be null");
     }
   }
 
diff --git a/hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java b/hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java
index 40db67b50..863103ebf 100644
--- a/hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/avro/TestHoodieAvroUtils.java
@@ -207,4 +207,34 @@ public class TestHoodieAvroUtils {
     Schema schemaWithoutMetaCols = HoodieAvroUtils.removeMetadataFields(schemaWithMetaCols);
     assertEquals(schemaWithoutMetaCols.getFields().size(), NUM_FIELDS_IN_EXAMPLE_SCHEMA);
   }
+
+  @Test
+  public void testGetNestedFieldVal() {
+    GenericRecord rec = new GenericData.Record(new Schema.Parser().parse(EXAMPLE_SCHEMA));
+    rec.put("_row_key", "key1");
+    rec.put("non_pii_col", "val1");
+    rec.put("pii_col", "val2");
+
+    Object rowKey = HoodieAvroUtils.getNestedFieldVal(rec, "_row_key", true);
+    assertEquals(rowKey, "key1");
+
+    Object rowKeyNotExist = HoodieAvroUtils.getNestedFieldVal(rec, "fake_key", true);
+    assertNull(rowKeyNotExist);
+
+    // Field does not exist
+    try {
+      HoodieAvroUtils.getNestedFieldVal(rec, "fake_key", false);
+    } catch (Exception e) {
+      assertEquals("fake_key(Part -fake_key) field not found in record. Acceptable fields were :[timestamp, _row_key, non_pii_col, pii_col]",
+          e.getMessage());
+    }
+
+    // Field exist while value not
+    try {
+      HoodieAvroUtils.getNestedFieldVal(rec, "timestamp", false);
+    } catch (Exception e) {
+      assertEquals("The value of timestamp can not be null", e.getMessage());
+    }
+  }
+
 }