@Override
    public SourceReader<Long, NumberSequenceSplit> createReader(SourceReaderContext readerContext) {
        return new IteratorSourceReader<>(readerContext);
    }
@Test
    void testReaderCheckpoints() throws Exception {
        final long from = 177;
        final long mid = 333;
        final long to = 563;
        final long elementsPerCycle = (to - from) / 3;

        final TestingReaderOutput<Long> out = new TestingReaderOutput<>();

        SourceReader<Long, NumberSequenceSource.NumberSequenceSplit> reader = createReader();
        reader.addSplits(
                Arrays.asList(
                        new NumberSequenceSource.NumberSequenceSplit("split-1", from, mid),
                        new NumberSequenceSource.NumberSequenceSplit("split-2", mid + 1, to)));

        long remainingInCycle = elementsPerCycle;
        while (reader.pollNext(out) != InputStatus.END_OF_INPUT) {
            if (--remainingInCycle <= 0) {
                remainingInCycle = elementsPerCycle;
                // checkpoint
                List<NumberSequenceSource.NumberSequenceSplit> splits = reader.snapshotState(1L);

                // re-create and restore
                reader = createReader();
                if (splits.isEmpty()) {
                    reader.notifyNoMoreSplits();
                } else {
                    reader.addSplits(splits);
                }
            }
        }

        final List<Long> result = out.getEmittedRecords();
        validateSequence(result, from, to);
    }