@SuppressWarnings("unchecked")
    public SplitT toFileSourceSplit() {
        final CheckpointedPosition position =
                (offset == CheckpointedPosition.NO_OFFSET && recordsToSkipAfterOffset == 0)
                        ? null
                        : new CheckpointedPosition(offset, recordsToSkipAfterOffset);

        final FileSourceSplit updatedSplit = split.updateWithCheckpointedPosition(position);

        // some sanity checks to avoid surprises and not accidentally lose split information
        if (updatedSplit == null) {
            throw new FlinkRuntimeException(
                    "Split returned 'null' in updateWithCheckpointedPosition(): " + split);
        }
        if (updatedSplit.getClass() != split.getClass()) {
            throw new FlinkRuntimeException(
                    String.format(
                            "Split returned different type in updateWithCheckpointedPosition(). "
                                    + "Split type is %s, returned type is %s",
                            split.getClass().getName(), updatedSplit.getClass().getName()));
        }

        return (SplitT) updatedSplit;
    }
@Test
    void testRoundTripWithoutModification() {
        final FileSourceSplit split = getTestSplit();
        final FileSourceSplitState state = new FileSourceSplitState(split);

        final FileSourceSplit resultSplit = state.toFileSourceSplit();

        assertThat(resultSplit.getReaderPosition()).isEqualTo(split.getReaderPosition());
    }