@Test
public void checkRestoredCheckpointWhenFetcherNotReady() throws Exception {

    @SuppressWarnings("unchecked") final FlinkPulsarSource<String> source = new DummyFlinkPulsarSource<>();
    final TestingListState<Tuple2<String, MessageId>> restoredListState = new TestingListState<>();
    setupSource(source, true, restoredListState, true, 0, 1);
    // snapshot before the fetcher starts running
    source.snapshotState(new StateSnapshotContextSynchronousImpl(17, 17));
    // ensure that the list was cleared and refilled. while this is an implementation detail, we use it here
    // to figure out that snapshotState() actually did something.
    Assert.assertTrue(restoredListState.isClearCalled());
    Set<Serializable> expected = new HashSet<>();
    for (Serializable serializable : restoredListState.get()) {
        expected.add(serializable);
    }
    int counter = 0;
    for (Serializable serializable : restoredListState.get()) {
        assertTrue(expected.contains(serializable));
        counter++;
    }
    assertEquals(expected.size(), counter);
}