@Test
public void checkRestoredCheckpointWhenFetcherNotReady() throws Exception {

	@SuppressWarnings("unchecked")
	final FlinkKafkaConsumerBase<String> consumer = new DummyFlinkKafkaConsumer<>();
	final TestingListState<Tuple2<KafkaTopicPartition, Long>> restoredListState = new TestingListState<>();
	setupConsumer(consumer, true, restoredListState, true, 0, 1);
	// snapshot before the fetcher starts running
	consumer.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);
}