Populate methods now always have 'customization' argument because

deserialize-method(s) of nested records may generate another
populate methods that require 'customization' (at least for compilation).

TDD approach - this commit adds unit test which fails and shows
where the issue actually is.

fastserde/avro-fastserde-tests-common/src/test/avro/recordWith2Fields.avsc

@@ -0,0 +1,17 @@
+{
+  "type": "record",
+  "name": "RecordWithOneNullableText",
+  "namespace": "com.linkedin.avro.fastserde.generated.avro",
+  "doc": "Used in tests of fast-serde to verify populate-methods works correctly with DatumReaderCustomization.",
+  "fields": [
+    {
+      "name": "text",
+      "type": [
+        "null",
+        "string"
+      ],
+      "default": null,
+      "doc": "Corresponds with recordWith2FieldsAndDeeplyNestedRecord.avsc"
+    }
+  ]
+}

fastserde/avro-fastserde-tests-common/src/test/avro/recordWith2FieldsAndDeeplyNestedRecord.avsc

@@ -0,0 +1,59 @@
+{
+  "type": "record",
+  "name": "RecordWithOneNullableTextAndDeeplyNestedRecord",
+  "namespace": "com.linkedin.avro.fastserde.generated.avro",
+  "doc": "Used in tests of fast-serde to verify populate-methods works correctly with DatumReaderCustomization. Just like OuterRecordWith2NestedBetaRecords but with one field more.",
+  "fields": [
+    {
+      "name": "text",
+      "type": [
+        "null",
+        "string"
+      ],
+      "default": null,
+      "doc": "Corresponds with recordWith2Fields.avsc"
+    },
+    {
+      "name": "nestedField",
+      "type": [
+        "null",
+        {
+          "name": "NestedRecord",
+          "type": "record",
+          "fields": [
+            {
+              "name": "sampleText1",
+              "type": [
+                "null",
+                "string"
+              ],
+              "default": null,
+              "doc": "field just to make crowd and force FastDeserializerGenerator to create populate*() method"
+            },
+            {
+              "name": "deeplyNestedField",
+              "type": [
+                "null",
+                {
+                  "name": "DeeplyNestedRecord",
+                  "type": "record",
+                  "fields": [
+                    {
+                      "name": "deeplyDeeplyNestedText",
+                      "type": [
+                        "null",
+                        "string"
+                      ],
+                      "default": null
+                    }
+                  ]
+                }
+              ]
+            }
+          ]
+        }
+      ],
+      "default": null
+    }
+  ]
+}

fastserde/avro-fastserde-tests-common/src/test/java/com/linkedin/avro/fastserde/FastSpecificDeserializerGeneratorTest.java

@@ -9,6 +9,8 @@
 import com.linkedin.avro.fastserde.generated.avro.OuterRecordWithNestedNotNullComplexFields;
 import com.linkedin.avro.fastserde.generated.avro.OuterRecordWithNestedNullableComplexFields;
 import com.linkedin.avro.fastserde.generated.avro.RecordWithLargeUnionField;
+import com.linkedin.avro.fastserde.generated.avro.RecordWithOneNullableText;
+import com.linkedin.avro.fastserde.generated.avro.RecordWithOneNullableTextAndDeeplyNestedRecord;
 import com.linkedin.avro.fastserde.generated.avro.RemovedTypesTestRecord;
 import com.linkedin.avro.fastserde.generated.avro.SplitRecordTest1;
 import com.linkedin.avro.fastserde.generated.avro.SplitRecordTest2;
@@ -46,6 +48,7 @@
 import org.testng.Assert;
 import org.testng.annotations.BeforeTest;
 import org.testng.annotations.DataProvider;
+import org.testng.annotations.Ignore;
 import org.testng.annotations.Test;
 import org.testng.collections.Lists;
 import org.testng.internal.collections.Pair;
@@ -110,7 +113,7 @@ public void prepare() throws Exception {
     classLoader = URLClassLoader.newInstance(new URL[]{tempDir.toURI().toURL()},
         FastSpecificDeserializerGeneratorTest.class.getClassLoader());
 
-    // In order to test the functionallity of the record split we set an unusually low number
+    // In order to test the functionality of the record split we set an unusually low number
     FastGenericDeserializerGenerator.setFieldsPerPopulationMethod(2);
   }
 
@@ -879,6 +882,35 @@ void deserializeNullableFieldsPreviouslySerializedAsNotNull(boolean useFastSeria
     Assert.assertEquals(outerRecord2.toString(), outerRecord1.toString());
   }
 
+  @Ignore
+  @Test(groups = {"deserializationTest"})
+  void deserializeWithSchemaMissingDeeplyNestedRecord() throws IOException {
+    // duplicates prepare() just in case - .avsc files used here assume FIELDS_PER_POPULATION_METHOD is 2
+    FastDeserializerGenerator.setFieldsPerPopulationMethod(2);
+
+    // given (serialized record with more fields than we want to read)
+    RecordWithOneNullableTextAndDeeplyNestedRecord reachRecord = new RecordWithOneNullableTextAndDeeplyNestedRecord();
+    setField(reachRecord, "text", "I am from reach record");
+
+    Schema writerSchema = RecordWithOneNullableTextAndDeeplyNestedRecord.SCHEMA$;
+    Schema readerSchema = RecordWithOneNullableText.SCHEMA$;
+
+    SpecificDatumWriter<RecordWithOneNullableTextAndDeeplyNestedRecord> datumWriter = new SpecificDatumWriter<>(writerSchema);
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    BinaryEncoder binaryEncoder = AvroCompatibilityHelper.newBinaryEncoder(baos);
+    datumWriter.write(reachRecord, binaryEncoder);
+    binaryEncoder.flush();
+
+    byte[] serializedReachRecord = baos.toByteArray();
+
+    // when (serialized reach record is read with schema without 'nestedField')
+    BinaryDecoder decoder = AvroCompatibilityHelper.newBinaryDecoder(serializedReachRecord);
+    RecordWithOneNullableText liteRecord = decodeRecordFast(readerSchema, writerSchema, decoder);
+
+    Assert.assertNotNull(liteRecord);
+    Assert.assertEquals(getField(liteRecord, "text").toString(), "I am from reach record");
+  }
+
   /**
    * @return serialized {@link OuterRecordWithNestedNotNullComplexFields}
    */

fastserde/avro-fastserde/src/main/java/com/linkedin/avro/fastserde/FastDeserializerGenerator.java

@@ -268,8 +268,10 @@ private void processRecord(JVar recordSchemaVar, String recordName, final Schema
     if (methodAlreadyDefined(recordWriterSchema, effectiveRecordReaderSchema, recordAction.getShouldRead())) {
       JMethod method = getMethod(recordWriterSchema, effectiveRecordReaderSchema, recordAction.getShouldRead());
       updateActualExceptions(method);
-      JExpression readingExpression = JExpr.invoke(method).arg(reuseSupplier.get()).arg(JExpr.direct(DECODER)).arg(
-          customizationSupplier.get());
+      JExpression readingExpression = JExpr.invoke(method)
+              .arg(reuseSupplier.get())
+              .arg(JExpr.direct(DECODER))
+              .arg(customizationSupplier.get());
       if (recordAction.getShouldRead()) {
         putRecordIntoParent.accept(parentBody, readingExpression);
       } else {
@@ -304,9 +306,15 @@ private void processRecord(JVar recordSchemaVar, String recordName, final Schema
     schemaAssistant.resetExceptionsFromStringable();
 
     if (recordAction.getShouldRead()) {
-      putRecordIntoParent.accept(parentBody, JExpr.invoke(method).arg(reuseSupplier.get()).arg(JExpr.direct(DECODER)).arg(customizationSupplier.get()));
+      putRecordIntoParent.accept(parentBody, JExpr.invoke(method)
+              .arg(reuseSupplier.get())
+              .arg(JExpr.direct(DECODER))
+              .arg(customizationSupplier.get()));
     } else {
-      parentBody.invoke(method).arg(reuseSupplier.get()).arg(JExpr.direct(DECODER)).arg(customizationSupplier.get());
+      parentBody.invoke(method)
+              .arg(reuseSupplier.get())
+              .arg(JExpr.direct(DECODER))
+              .arg(customizationSupplier.get());
     }
 
     JBlock methodBody = method.body();