[da-vinci]Log more informative details in UpdateBuilder using unresol…

…vedDatum object (#722)

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

@@ -13,6 +13,8 @@
 import java.util.Map;
 import java.util.Set;
 import org.apache.avro.Schema;
+import org.apache.avro.UnresolvedUnionException;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.commons.lang3.Validate;
@@ -127,9 +129,17 @@ public GenericRecord build() {
     return updateRecord;
   }
 
-  private Exception validateUpdateRecordIsSerializable(GenericRecord updateRecord) {
+  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();
+      return new VeniceException(
+          "The following type does not conform to any branch of the union: " + unresolvedDatumType,
+          serializationException);
     } catch (Exception serializationException) {
       return serializationException;
     }

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

@@ -346,6 +346,19 @@ public void testSetFieldToNull() {
     Assert.assertThrows(VeniceException.class, () -> builder2.setNewFieldValue("name", null));
   }
 
+  /**
+   * Should return a VeniceException with more information on the unresolved datum type when field is of type union
+   * and the value doesn't match one of the union branches.
+   */
+  @Test
+  public void testValidateUpdateRecordIsSerializable() {
+    GenericRecord record = new GenericData.Record(UPDATE_SCHEMA);
+    UpdateBuilderImpl builder = new UpdateBuilderImpl(UPDATE_SCHEMA);
+    record.put("name", true);
+    Exception e = builder.validateUpdateRecordIsSerializable(record);
+    Assert.assertTrue(e instanceof VeniceException);
+  }
+
   private GenericRecord createRecordForListField(int number) {
     Schema recordSchema = VALUE_SCHEMA.getField("recordArray").schema().getElementType();
     GenericRecord record = new GenericData.Record(recordSchema);