[producer] Fixed NPE in UpdateBuilderImpl::validateUpdateRecordIsSeri…

…alizable (#740)

Also expanded the unit test.

internal/venice-common/src/main/java/com/linkedin/venice/writer/update/UpdateBuilderImpl.java

@@ -133,23 +133,30 @@ Exception validateUpdateRecordIsSerializable(GenericRecord updateRecord) {
     try {
       serializer.serialize(updateRecord);
     } catch (UnresolvedUnionException serializationException) {
-      Object unresolvedDatum = serializationException.getUnresolvedDatum();
-      String unresolvedDatumType = unresolvedDatum instanceof GenericContainer
-          ? ((GenericContainer) unresolvedDatum).getSchema().toString()
-          : unresolvedDatum.getClass().getSimpleName();
+      String datumDescription = datumDescription(serializationException.getUnresolvedDatum());
       return new VeniceException(
-          "The following type does not conform to any branch of the union: " + unresolvedDatumType,
+          "The following type does not conform to any branch of the union: " + datumDescription,
           serializationException);
     } catch (Exception serializationException) {
       return serializationException;
     }
     return null;
   }
 
+  private String datumDescription(Object unresolvedDatum) {
+    if (unresolvedDatum instanceof GenericContainer) {
+      return ((GenericContainer) unresolvedDatum).getSchema().toString();
+    } else if (unresolvedDatum == null) {
+      return "null";
+    } else {
+      return unresolvedDatum.getClass().getSimpleName();
+    }
+  }
+
   /**
    * Given a field from the partial update schema and find the schema of its corresponding value field.
    *
-   * @param updateField A field from the partial update schema.
+   * @param updateFieldName A field from the partial update schema.
    * @return Schema of its corresponding value field.
    */
   private Schema getCorrespondingValueFieldSchema(String updateFieldName) {

internal/venice-common/src/test/java/com/linkedin/venice/writer/update/UpdateBuilderImplTest.java

@@ -1,5 +1,9 @@
 package com.linkedin.venice.writer.update;
 
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
 import com.linkedin.alpini.io.IOUtils;
 import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper;
 import com.linkedin.venice.exceptions.VeniceException;
@@ -354,9 +358,42 @@ public void testSetFieldToNull() {
   public void testValidateUpdateRecordIsSerializable() {
     GenericRecord record = new GenericData.Record(UPDATE_SCHEMA);
     UpdateBuilderImpl builder = new UpdateBuilderImpl(UPDATE_SCHEMA);
-    record.put("name", true);
+
+    // The "name" field is mandatory, so not setting it will leave it null, which is wrong.
     Exception e = builder.validateUpdateRecordIsSerializable(record);
-    Assert.assertTrue(e instanceof VeniceException);
+    assertTrue(e instanceof VeniceException);
+    assertEquals(e.getMessage(), "The following type does not conform to any branch of the union: null");
+
+    // The "name" field is of type string, so a boolean value is also wrong
+    record.put("name", true);
+    e = builder.validateUpdateRecordIsSerializable(record);
+    assertTrue(e instanceof VeniceException);
+    assertEquals(e.getMessage(), "The following type does not conform to any branch of the union: Boolean");
+
+    // All good values
+    record.put("name", "John Doe");
+    record.put("age", 60);
+    // The "address" field can be omitted since it allows null...
+    record.put("intArray", Collections.emptyList());
+    record.put("recordArray", Collections.emptyList());
+    record.put("stringMap", Collections.emptyMap());
+    record.put("recordMap", Collections.emptyMap());
+    e = builder.validateUpdateRecordIsSerializable(record);
+    assertNull(e);
+
+    // Almost correct record, but with wrong namespace
+    List<Schema.Field> fields = new ArrayList<>(2);
+    fields.add(
+        AvroCompatibilityHelper.createSchemaField("streetaddress", Schema.create(Schema.Type.STRING), "", "unknown"));
+    fields.add(AvroCompatibilityHelper.createSchemaField("city", Schema.create(Schema.Type.STRING), "", "Sunnyvale"));
+    Schema wrongAddressSchema = Schema.createRecord("AddressUSRecord", "", "wrong.namespace", false, fields);
+    GenericRecord wrongAddressRecord = new GenericData.Record(wrongAddressSchema);
+    record.put("address", wrongAddressRecord);
+    e = builder.validateUpdateRecordIsSerializable(record);
+    assertTrue(e instanceof VeniceException);
+    assertEquals(
+        e.getMessage(),
+        "The following type does not conform to any branch of the union: {\"type\":\"record\",\"name\":\"AddressUSRecord\",\"namespace\":\"wrong.namespace\",\"doc\":\"\",\"fields\":[{\"name\":\"streetaddress\",\"type\":\"string\",\"doc\":\"\",\"default\":\"unknown\"},{\"name\":\"city\",\"type\":\"string\",\"doc\":\"\",\"default\":\"Sunnyvale\"}]}");
   }
 
   private GenericRecord createRecordForListField(int number) {