flat-record: objectMapper.readFlat can parse an enum that had custom properties set on it (#630)

Author: eduardoramirez

hollow/src/main/java/com/netflix/hollow/core/write/objectmapper/HollowObjectMapper.java

@@ -94,8 +94,9 @@ public <T> T readFlat(FlatRecord record) {
                 recordReader.skipSchema(schema);
             } else {
                 Object obj = mapper.parseFlatRecord(schema, recordReader, parsedObjects);
-                parsedObjects.put(ordinal++, obj);
+                parsedObjects.put(ordinal, obj);
             }
+            ordinal++;
         }
 
         return (T) parsedObjects.get(ordinal - 1);

hollow/src/main/java/com/netflix/hollow/core/write/objectmapper/HollowObjectTypeMapper.java

@@ -195,7 +195,7 @@ protected Object parseFlatRecord(HollowSchema recordSchema, FlatRecordReader rea
             HollowObjectSchema recordObjectSchema = (HollowObjectSchema) recordSchema;
 
             Object obj = null;
-            if (BOXED_WRAPPERS.contains(clazz) || clazz.isEnum()) {
+            if (BOXED_WRAPPERS.contains(clazz)) {
                 // if `clazz` is a BoxedWrapper then by definition its OBJECT schema will have a single primitive
                 // field so find it in the FlatRecord and ignore all other fields.
                 for (int i = 0; i < recordObjectSchema.numFields(); i++) {
@@ -206,7 +206,20 @@ protected Object parseFlatRecord(HollowSchema recordSchema, FlatRecordReader rea
                         reader.skipField(recordObjectSchema.getFieldType(i));
                     }
                 }
-            } else  {
+            } else if (clazz.isEnum()) {
+                // if `clazz` is an enum, then we should expect to find a field called `_name` in the FlatRecord.
+                // There may be other fields if the producer enum contained custom properties, we ignore them
+                // here assuming the enum constructor will set them if needed.
+                for (int i = 0; i < recordObjectSchema.numFields(); i++) {
+                    String fieldName = recordObjectSchema.getFieldName(i);
+                    int posInPojoSchema = schema.getPosition(fieldName);
+                    if (fieldName.equals(MappedFieldType.ENUM_NAME.getSpecialFieldName()) && posInPojoSchema != -1) {
+                        obj = mappedFields.get(posInPojoSchema).parseBoxedWrapper(reader);
+                    } else {
+                        reader.skipField(recordObjectSchema.getFieldType(i));
+                    }
+                }
+            } else {
                 obj = unsafe.allocateInstance(clazz);
                 for (int i = 0; i < recordObjectSchema.numFields(); i++) {
                     int posInPojoSchema = schema.getPosition(recordObjectSchema.getFieldName(i));

hollow/src/test/java/com/netflix/hollow/core/write/objectmapper/HollowObjectMapperFlatRecordParserTest.java

@@ -38,6 +38,7 @@ public void testSpecialWrapperTypes() {
     SpecialWrapperTypesTest wrapperTypesTest = new SpecialWrapperTypesTest();
     wrapperTypesTest.id = 8797182L;
     wrapperTypesTest.type = AnEnum.SOME_VALUE_C;
+    wrapperTypesTest.complexEnum = ComplexEnum.SOME_VALUE_A;
     wrapperTypesTest.dateCreated = new Date();
 
     flatRecordWriter.reset();
@@ -47,6 +48,8 @@ public void testSpecialWrapperTypes() {
     SpecialWrapperTypesTest result = mapper.readFlat(fr);
 
     Assert.assertEquals(wrapperTypesTest, result);
+    Assert.assertEquals(wrapperTypesTest.complexEnum.value, result.complexEnum.value);
+    Assert.assertEquals(wrapperTypesTest.complexEnum.anotherValue, result.complexEnum.anotherValue);
   }
 
   @Test
@@ -558,19 +561,35 @@ enum AnEnum {
     SOME_VALUE_C,
   }
 
+  enum ComplexEnum {
+    SOME_VALUE_A("A", 1),
+    SOME_VALUE_B("B", 2),
+    SOME_VALUE_C("C", 3);
+
+    final String value;
+    final int anotherValue;
+
+    ComplexEnum(String value, int anotherValue) {
+      this.value = value;
+      this.anotherValue = anotherValue;
+    }
+  }
+
   @HollowTypeName(name = "SpecialWrapperTypesTest")
   @HollowPrimaryKey(fields = {"id"})
   static class SpecialWrapperTypesTest {
     long id;
     @HollowTypeName(name = "AnEnum")
     AnEnum type;
+    @HollowTypeName(name = "ComplexEnum")
+    ComplexEnum complexEnum;
     Date dateCreated;
 
     @Override
     public boolean equals(Object o) {
       if(o instanceof SpecialWrapperTypesTest) {
         SpecialWrapperTypesTest other = (SpecialWrapperTypesTest)o;
-        return id == other.id && type == other.type && dateCreated.equals(other.dateCreated);
+        return id == other.id && complexEnum == other.complexEnum && type == other.type && dateCreated.equals(other.dateCreated);
       }
       return false;
     }
@@ -580,6 +599,7 @@ public String toString() {
       return "SpecialWrapperTypesTest{" +
               "id=" + id +
               ", type='" + type + '\'' +
+              ", complexEnum='" + complexEnum + '\'' +
               ", dateCreated=" + dateCreated +
               '}';
     }