### Question:
PendingCheckpointStats extends AbstractCheckpointStats { void reportFailedCheckpoint(long failureTimestamp, @Nullable Throwable cause) { FailedCheckpointStats failed = new FailedCheckpointStats( checkpointId, triggerTimestamp, props, numberOfSubtasks, new HashMap<>(taskStats), currentNumAcknowledgedSubtasks, currentStateSize, currentAlignmentBuffered, failureTimestamp, latestAcknowledgedSubtask, cause); trackerCallback.reportFailedCheckpoint(failed); } PendingCheckpointStats(
			long checkpointId,
			long triggerTimestamp,
			CheckpointProperties props,
			int totalSubtaskCount,
			Map<JobVertexID, TaskStateStats> taskStats,
			CheckpointStatsTracker.PendingCheckpointStatsCallback trackerCallback); @Override CheckpointStatsStatus getStatus(); @Override int getNumberOfAcknowledgedSubtasks(); @Override long getStateSize(); @Override long getAlignmentBuffered(); @Override SubtaskStateStats getLatestAcknowledgedSubtaskStats(); @Override String toString();  }### Answer:
@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.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION), task1.getNumberOfSubtasks() + task2.getNumberOfSubtasks(), taskStats, callback); 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)); } 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())); }