@Override
    public PendingSplitsCheckpoint<FileSourceSplit> snapshotState(long checkpointId)
            throws Exception {
        final PendingSplitsCheckpoint<FileSourceSplit> checkpoint =
                PendingSplitsCheckpoint.fromCollectionSnapshot(
                        splitAssigner.remainingSplits(), pathsAlreadyProcessed);

        LOG.debug("Source Checkpoint is {}", checkpoint);
        return checkpoint;
    }
@Test
    void testDiscoverSplitWhenNoReaderRegistered() throws Exception {
        final TestingFileEnumerator fileEnumerator = new TestingFileEnumerator();
        final TestingSplitEnumeratorContext<FileSourceSplit> context =
                new TestingSplitEnumeratorContext<>(4);
        final ContinuousFileSplitEnumerator enumerator = createEnumerator(fileEnumerator, context);

        // make one split available and trigger the periodic discovery
        final FileSourceSplit split = createRandomSplit();
        fileEnumerator.addSplits(split);
        context.triggerAllActions();

        assertThat(enumerator.snapshotState(1L).getSplits()).contains(split);
    }