@Test
public void testReportFailedCheckpoint() throws Exception {

	TaskStateStats task1 = new TaskStateStats(new JobVertexID(), 3);
	TaskStateStats task2 = new TaskStateStats(new JobVertexID(), 4);
	HashMap<JobVertexID, TaskStateStats> taskStats = new HashMap<>();
	taskStats.put(task1.getJobVertexId(), task1);
	taskStats.put(task2.getJobVertexId(), task2);
	CheckpointStatsTracker.PendingCheckpointStatsCallback callback = mock(
		CheckpointStatsTracker.PendingCheckpointStatsCallback.class);
	long triggerTimestamp = 123123;
	PendingCheckpointStats pending = new PendingCheckpointStats(
		0,
		triggerTimestamp,
		CheckpointProperties.forStandardCheckpoint(),
		task1.getNumberOfSubtasks() + task2.getNumberOfSubtasks(),
		taskStats,
		callback);
	// Report subtasks
	for (int i = 0; i < task1.getNumberOfSubtasks(); i++) {
		pending.reportSubtaskStats(task1.getJobVertexId(), createSubtaskStats(i));
	}
	for (int i = 0; i < task2.getNumberOfSubtasks(); i++) {
		pending.reportSubtaskStats(task2.getJobVertexId(), createSubtaskStats(i));
	}
	// Report failed
	Exception cause = new Exception("test exception");
	long failureTimestamp = 112211137;
	pending.reportFailedCheckpoint(failureTimestamp, cause);
	ArgumentCaptor<FailedCheckpointStats> args = ArgumentCaptor.forClass(FailedCheckpointStats.class);
	verify(callback).reportFailedCheckpoint(args.capture());
	FailedCheckpointStats failed = args.getValue();
	assertNotNull(failed);
	assertEquals(CheckpointStatsStatus.FAILED, failed.getStatus());
	assertEquals(failureTimestamp, failed.getFailureTimestamp());
	assertEquals(cause.getMessage(), failed.getFailureMessage());
	assertEquals(pending.getCheckpointId(), failed.getCheckpointId());
	assertEquals(pending.getNumberOfAcknowledgedSubtasks(), failed.getNumberOfAcknowledgedSubtasks());
	assertEquals(pending.getLatestAcknowledgedSubtaskStats(), failed.getLatestAcknowledgedSubtaskStats());
	assertEquals(pending.getLatestAckTimestamp(), failed.getLatestAckTimestamp());
	assertEquals(failureTimestamp - triggerTimestamp, failed.getEndToEndDuration());
	assertEquals(pending.getStateSize(), failed.getStateSize());
	assertEquals(pending.getAlignmentBuffered(), failed.getAlignmentBuffered());
	assertEquals(task1, failed.getTaskStateStats(task1.getJobVertexId()));
	assertEquals(task2, failed.getTaskStateStats(task2.getJobVertexId()));
}