@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 testCreateSpecificReader() throws IOException {
        StreamFormat.Reader<Address> reader =
                createReader(
                        AvroParquetReaders.forSpecificRecord(Address.class),
                        new Configuration(),
                        addressPath,
                        0,
                        addressPath.getFileSystem().getFileStatus(addressPath).getLen());
        for (Address expected : addressRecords) {
            Address address = Objects.requireNonNull(reader.read());
            assertThat(address).isEqualTo(expected);
        }
    }