@Override
    public BulkFormat.Reader<T> createReader(
            final Configuration config, final FileSourceSplit split) throws IOException {

        final TrackingFsDataInputStream trackingStream =
                openStream(split.path(), config, split.offset());
        final long splitEnd = split.offset() + split.length();

        return doWithCleanupOnException(
                trackingStream,
                () -> {
                    final StreamFormat.Reader<T> streamReader =
                            streamFormat.createReader(
                                    config,
                                    trackingStream,
                                    trackingStream.getFileLength(),
                                    splitEnd);
                    return new Reader<>(
                            streamReader, trackingStream, CheckpointedPosition.NO_OFFSET, 0L);
                });
    }
@Test
    void testReadEmptyFile() throws IOException {
        final StreamFormatAdapter<Integer> format =
                new StreamFormatAdapter<>(new CheckpointedIntFormat());

        final File emptyFile = new File(tmpDir.toFile(), "testFile-empty");
        emptyFile.createNewFile();
        Path emptyFilePath = Path.fromLocalFile(emptyFile);

        final BulkFormat.Reader<Integer> reader =
                format.createReader(
                        new Configuration(),
                        new FileSourceSplit("test-id", emptyFilePath, 0L, 0, 0L, 0));

        final List<Integer> result = new ArrayList<>();
        readNumbers(reader, result, 0);

        assertThat(result).isEmpty();
    }