@Test
public void testCompletedCheckpointReordering() throws Exception {

	CompletedCheckpoint[] checkpoints = generateRandomCheckpoints(2);
	ExecutionVertex[] tasksToWaitFor = createTasksToWaitFor(checkpoints[0]);
	CheckpointStatsTracker tracker = new SimpleCheckpointStatsTracker(10, tasksToWaitFor);
	// First the second checkpoint notifies
	tracker.onCompletedCheckpoint(checkpoints[1]);
	verifyJobStats(tracker, 10, new CompletedCheckpoint[] { checkpoints[1] });
	verifySubtaskStats(tracker, tasksToWaitFor, checkpoints[1]);
	// Then the first one
	tracker.onCompletedCheckpoint(checkpoints[0]);
	verifyJobStats(tracker, 10, checkpoints);
	// This should not alter the results for the subtasks
	verifySubtaskStats(tracker, tasksToWaitFor, checkpoints[1]);
}