@Override
    public PendingSplitsCheckpoint<FileSourceSplit> snapshotState(long checkpointId) {
        return PendingSplitsCheckpoint.fromCollectionSnapshot(splitAssigner.remainingSplits());
    }
@Test
    void testCheckpointNoSplitRequested() throws Exception {
        final TestingSplitEnumeratorContext<FileSourceSplit> context =
                new TestingSplitEnumeratorContext<>(4);
        final FileSourceSplit split = createRandomSplit();
        final StaticFileSplitEnumerator enumerator = createEnumerator(context, split);

        final PendingSplitsCheckpoint<FileSourceSplit> checkpoint = enumerator.snapshotState(1L);

        assertThat(checkpoint.getSplits()).contains(split);
    }