@Override
    public Reader<E> createReader(
            Configuration config, FSDataInputStream stream, long fileLen, long splitEnd)
            throws IOException {

        // current version does not support splitting.
        checkNotSplit(fileLen, splitEnd);

        return new AvroParquetRecordReader<E>(
                AvroParquetReader.<E>builder(new ParquetInputFile(stream, fileLen))
                        .withDataModel(getDataModel())
                        .withConf(HadoopUtils.getHadoopConfiguration(config))
                        .build());
    }
@Test
    void testCreateReflectReader() throws IOException {
        StreamFormat.Reader<Datum> reader =
                createReader(
                        AvroParquetReaders.forReflectRecord(Datum.class),
                        new Configuration(),
                        datumPath,
                        0,
                        datumPath.getFileSystem().getFileStatus(datumPath).getLen());
        for (Datum expected : datumRecords) {
            assertThat(reader.read()).isNotNull().isEqualTo(expected);
        }
    }