[Bug]: ParquetIO failing to read with specified schema #27234

java.lang.ArrayIndexOutOfBoundsException: 3
org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.ArrayIndexOutOfBoundsException: 3
    at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:371)
    at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:339)
    at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:219)
    at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:67)

String path = "pathToFile";
Schema schema = ReflectData.get().getSchema(EntityClass.class)

PCollection<GenericRecord> records = pipeline.apply("read a file", ParquetIO.read(schema).from(path));

pipeline.run().waitUntilFinish();

@With
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
@ParametersAreNonnullByDefault
@DefaultSchema(JavaBeanSchema.class)
public class EntityClass {
    
    private String firstName;
    
    private String middleName;
    
}    

@With
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
@ParametersAreNonnullByDefault
@DefaultSchema(JavaBeanSchema.class)
public class EntityClass {

    private String firstName;

    private String middleName;

    @Nullable
    @org.apache.avro.reflect.Nullable
    private String lastName;
}

{
    "type": "record",
    "name": "EntityClass",
    "namespace": "com.test.entityclass",
    "fields": [{
        "name": "firstName",
        "type": "string"
    }, {
        "name": "middleName",
        "type": "string"
    }, {
        "name": "lastName",
        "type": ["null", "string"],
        "default": null
    }]
}

AvroReadSupport.setAvroReadSchema(conf, NEW_SCHEMA);

try (ParquetReader<GenericRecord> reader = AvroParquetReader.<GenericRecord>builder(dataFile)
                .withConf(conf)
                .build()) {

GenericRecord genericRecord;
while ((genericRecord = reader.read()) != null) {
     //System.out.println(genericRecord.getSchema());
     T result = apply((GenericRecord) genericRecord.get("payload"), clazz);
     resultFiles.add(result);
   }
}