@Test
public void testCreateSnapshot() throws Exception {

	ExecutionJobVertex jobVertex = mock(ExecutionJobVertex.class);
	when(jobVertex.getJobVertexId()).thenReturn(new JobVertexID());
	when(jobVertex.getParallelism()).thenReturn(1);
	CheckpointStatsTracker tracker = new CheckpointStatsTracker(
		10,
		Collections.singletonList(jobVertex),
		mock(CheckpointCoordinatorConfiguration.class),
		new UnregisteredMetricsGroup());
	CheckpointStatsSnapshot snapshot1 = tracker.createSnapshot();
	// Pending checkpoint => new snapshot
	PendingCheckpointStats pending = tracker.reportPendingCheckpoint(
		0,
		1,
		CheckpointProperties.forStandardCheckpoint());
	pending.reportSubtaskStats(jobVertex.getJobVertexId(), createSubtaskStats(0));
	CheckpointStatsSnapshot snapshot2 = tracker.createSnapshot();
	assertNotEquals(snapshot1, snapshot2);
	assertEquals(snapshot2, tracker.createSnapshot());
	// Complete checkpoint => new snapshot
	pending.reportCompletedCheckpoint(null);
	CheckpointStatsSnapshot snapshot3 = tracker.createSnapshot();
	assertNotEquals(snapshot2, snapshot3);
	// Restore operation => new snapshot
	tracker.reportRestoredCheckpoint(new RestoredCheckpointStats(12, CheckpointProperties.forStandardCheckpoint(), 12, null));
	CheckpointStatsSnapshot snapshot4 = tracker.createSnapshot();
	assertNotEquals(snapshot3, snapshot4);
	assertEquals(snapshot4, tracker.createSnapshot());
}