SimpleCheckpointStatsTracker implements CheckpointStatsTracker { @Override public void onCompletedCheckpoint(CompletedCheckpoint checkpoint) { if (taskParallelism.isEmpty()) { return; } synchronized (statsLock) { long overallStateSize = 0; Map<JobVertexID, long[][]> statsForSubTasks = new HashMap<>(); for (StateForTask state : checkpoint.getStates()) { overallStateSize += state.getStateSize(); JobVertexID opId = state.getOperatorId(); long[][] statsPerSubtask = statsForSubTasks.get(opId); if (statsPerSubtask == null) { int parallelism = taskParallelism.get(opId); statsPerSubtask = new long[parallelism][2]; statsForSubTasks.put(opId, statsPerSubtask); } int subTaskIndex = state.getSubtask(); if (subTaskIndex < statsPerSubtask.length) { statsPerSubtask[subTaskIndex][0] = state.getDuration(); statsPerSubtask[subTaskIndex][1] = state.getStateSize(); } } boolean isInOrder = latestCompletedCheckpoint != null && checkpoint.getCheckpointID() > latestCompletedCheckpoint.getCheckpointID(); lastJobStats = null; if (overallCount == 0 || isInOrder) { latestCompletedCheckpoint = checkpoint; operatorStatsCache.clear(); subTaskStats = statsForSubTasks; } long checkpointId = checkpoint.getCheckpointID(); long checkpointTriggerTimestamp = checkpoint.getTimestamp(); long checkpointDuration = checkpoint.getDuration(); overallCount++; if (checkpointDuration > overallMaxDuration) { overallMaxDuration = checkpointDuration; } if (checkpointDuration < overallMinDuration) { overallMinDuration = checkpointDuration; } overallTotalDuration += checkpointDuration; if (overallStateSize < overallMinStateSize) { overallMinStateSize = overallStateSize; } if (overallStateSize > overallMaxStateSize) { overallMaxStateSize = overallStateSize; } this.overallTotalStateSize += overallStateSize; if (historySize > 0) { CheckpointStats stats = new CheckpointStats( checkpointId, checkpointTriggerTimestamp, checkpointDuration, overallStateSize); if (isInOrder) { if (history.size() == historySize) { history.remove(0); } history.add(stats); } else { final int size = history.size(); if (size == historySize) { if (checkpointId > history.get(0).getCheckpointId()) { history.remove(0); } } int pos = 0; for (int i = 0; i < size; i++) { pos = i; if (checkpointId < history.get(i).getCheckpointId()) { break; } } history.add(pos, stats); } } } } SimpleCheckpointStatsTracker(
			int historySize,
			ExecutionVertex[] tasksToWaitFor); @Override void onCompletedCheckpoint(CompletedCheckpoint checkpoint); @SuppressWarnings("unchecked") @Override Option<JobCheckpointStats> getJobStats(); @Override Option<OperatorCheckpointStats> getOperatorStats(JobVertexID operatorId);  }
@Test public void testRandomStats() throws Exception { CompletedCheckpoint[] checkpoints = generateRandomCheckpoints(16); ExecutionVertex[] tasksToWaitFor = createTasksToWaitFor(checkpoints[0]); CheckpointStatsTracker tracker = new SimpleCheckpointStatsTracker(10, tasksToWaitFor); for (int i = 0; i < checkpoints.length; i++) { CompletedCheckpoint checkpoint = checkpoints[i]; tracker.onCompletedCheckpoint(checkpoint); verifyJobStats(tracker, 10, Arrays.copyOfRange(checkpoints, 0, i + 1)); verifySubtaskStats(tracker, tasksToWaitFor, checkpoint); } }
@Test public void testCompletedCheckpointReordering() throws Exception { CompletedCheckpoint[] checkpoints = generateRandomCheckpoints(2); ExecutionVertex[] tasksToWaitFor = createTasksToWaitFor(checkpoints[0]); CheckpointStatsTracker tracker = new SimpleCheckpointStatsTracker(10, tasksToWaitFor); tracker.onCompletedCheckpoint(checkpoints[1]); verifyJobStats(tracker, 10, new CompletedCheckpoint[] { checkpoints[1] }); verifySubtaskStats(tracker, tasksToWaitFor, checkpoints[1]); tracker.onCompletedCheckpoint(checkpoints[0]); verifyJobStats(tracker, 10, checkpoints); verifySubtaskStats(tracker, tasksToWaitFor, checkpoints[1]); }