test_class.declarator: RAND = new Random()
test_class.modifier: private static final
test_class.original_string: private static final Random RAND = new Random();
test_class.type: Random
test_class.var_name: RAND
test_class.file: flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/stats/SimpleCheckpointStatsTrackerTest.java
test_class.identifier: SimpleCheckpointStatsTrackerTest
test_case.body: @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_case.class_method_signature: SimpleCheckpointStatsTrackerTest.testRandomStats()
test_case.full_signature: @Test public void testRandomStats()
test_case.identifier: testRandomStats
test_case.invocations: generateRandomCheckpoints createTasksToWaitFor onCompletedCheckpoint verifyJobStats copyOfRange verifySubtaskStats
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testRandomStats()
focal_class.declarator: statsLock = new Object()
focal_class.modifier: private final
focal_class.original_string: private final Object statsLock = new Object();
focal_class.type: Object
focal_class.var_name: statsLock
focal_class.declarator: historySize
focal_class.modifier: private final
focal_class.original_string: private final int historySize;
focal_class.type: int
focal_class.var_name: historySize
focal_class.declarator: history = new ArrayList<>()
focal_class.modifier: private final
focal_class.original_string: private final ArrayList<CheckpointStats> history = new ArrayList<>();
focal_class.type: ArrayList<CheckpointStats>
focal_class.var_name: history
focal_class.declarator: taskParallelism
focal_class.modifier: private final
focal_class.original_string: private final Map<JobVertexID, Integer> taskParallelism;
focal_class.type: Map<JobVertexID, Integer>
focal_class.var_name: taskParallelism
focal_class.declarator: subTaskStats
focal_class.modifier: private
focal_class.original_string: private Map<JobVertexID, long[][]> subTaskStats;
focal_class.type: Map<JobVertexID, long[][]>
focal_class.var_name: subTaskStats
focal_class.declarator: lastJobStats
focal_class.modifier: private
focal_class.original_string: private JobCheckpointStats lastJobStats;
focal_class.type: JobCheckpointStats
focal_class.var_name: lastJobStats
focal_class.declarator: operatorStatsCache = new HashMap<>()
focal_class.modifier: private
focal_class.original_string: private Map<JobVertexID, OperatorCheckpointStats> operatorStatsCache = new HashMap<>();
focal_class.type: Map<JobVertexID, OperatorCheckpointStats>
focal_class.var_name: operatorStatsCache
focal_class.declarator: overallCount
focal_class.modifier: private
focal_class.original_string: private long overallCount;
focal_class.type: long
focal_class.var_name: overallCount
focal_class.declarator: overallMinDuration = Long.MAX_VALUE
focal_class.modifier: private
focal_class.original_string: private long overallMinDuration = Long.MAX_VALUE;
focal_class.type: long
focal_class.var_name: overallMinDuration
focal_class.declarator: overallMaxDuration = Long.MIN_VALUE
focal_class.modifier: private
focal_class.original_string: private long overallMaxDuration = Long.MIN_VALUE;
focal_class.type: long
focal_class.var_name: overallMaxDuration
focal_class.declarator: overallTotalDuration
focal_class.modifier: private
focal_class.original_string: private long overallTotalDuration;
focal_class.type: long
focal_class.var_name: overallTotalDuration
focal_class.declarator: overallMinStateSize = Long.MAX_VALUE
focal_class.modifier: private
focal_class.original_string: private long overallMinStateSize = Long.MAX_VALUE;
focal_class.type: long
focal_class.var_name: overallMinStateSize
focal_class.declarator: overallMaxStateSize = Long.MIN_VALUE
focal_class.modifier: private
focal_class.original_string: private long overallMaxStateSize = Long.MIN_VALUE;
focal_class.type: long
focal_class.var_name: overallMaxStateSize
focal_class.declarator: overallTotalStateSize
focal_class.modifier: private
focal_class.original_string: private long overallTotalStateSize;
focal_class.type: long
focal_class.var_name: overallTotalStateSize
focal_class.declarator: latestCompletedCheckpoint
focal_class.modifier: private
focal_class.original_string: private CompletedCheckpoint latestCompletedCheckpoint;
focal_class.type: CompletedCheckpoint
focal_class.var_name: latestCompletedCheckpoint
focal_class.file: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/stats/SimpleCheckpointStatsTracker.java
focal_class.identifier: SimpleCheckpointStatsTracker
focal_class.interfaces: implements CheckpointStatsTracker
focal_class.class_method_signature: SimpleCheckpointStatsTracker.SimpleCheckpointStatsTracker(
  			int historySize,
  			ExecutionVertex[] tasksToWaitFor)
focal_class.full_signature: public  SimpleCheckpointStatsTracker(
  			int historySize,
  			ExecutionVertex[] tasksToWaitFor)
focal_class.identifier: SimpleCheckpointStatsTracker
focal_class.modifiers: public
focal_class.parameters: (
  			int historySize,
  			ExecutionVertex[] tasksToWaitFor)
focal_class.signature:  SimpleCheckpointStatsTracker(
  			int historySize,
  			ExecutionVertex[] tasksToWaitFor)
focal_class.class_method_signature: SimpleCheckpointStatsTracker.onCompletedCheckpoint(CompletedCheckpoint checkpoint)
focal_class.full_signature: @Override public void onCompletedCheckpoint(CompletedCheckpoint checkpoint)
focal_class.identifier: onCompletedCheckpoint
focal_class.modifiers: @Override public
focal_class.parameters: (CompletedCheckpoint checkpoint)
focal_class.return: void
focal_class.signature: void onCompletedCheckpoint(CompletedCheckpoint checkpoint)
focal_class.class_method_signature: SimpleCheckpointStatsTracker.getJobStats()
focal_class.full_signature: @SuppressWarnings("unchecked") @Override public Option<JobCheckpointStats> getJobStats()
focal_class.identifier: getJobStats
focal_class.modifiers: @SuppressWarnings("unchecked") @Override public
focal_class.parameters: ()
focal_class.return: Option<JobCheckpointStats>
focal_class.signature: Option<JobCheckpointStats> getJobStats()
focal_class.class_method_signature: SimpleCheckpointStatsTracker.getOperatorStats(JobVertexID operatorId)
focal_class.full_signature: @Override public Option<OperatorCheckpointStats> getOperatorStats(JobVertexID operatorId)
focal_class.identifier: getOperatorStats
focal_class.modifiers: @Override public
focal_class.parameters: (JobVertexID operatorId)
focal_class.return: Option<OperatorCheckpointStats>
focal_class.signature: Option<OperatorCheckpointStats> getOperatorStats(JobVertexID operatorId)
focal_method.body: @Override
  	public void onCompletedCheckpoint(CompletedCheckpoint checkpoint) {
  		// Sanity check
  		if (taskParallelism.isEmpty()) {
  			return;
  		}
  
  		synchronized (statsLock) {
  			long overallStateSize = 0;
  
  			// Operator stats
  			Map<JobVertexID, long[][]> statsForSubTasks = new HashMap<>();
  
  			for (StateForTask state : checkpoint.getStates()) {
  				// Job-level checkpoint size is sum of all state sizes
  				overallStateSize += state.getStateSize();
  
  				// Subtask stats
  				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();
  				}
  			}
  
  			// It is possible that completed checkpoints are added out of
  			// order. Make sure that in this case the last completed
  			// checkpoint is not updated.
  			boolean isInOrder = latestCompletedCheckpoint != null &&
  					checkpoint.getCheckpointID() > latestCompletedCheckpoint.getCheckpointID();
  
  			// Clear this in each case
  			lastJobStats = null;
  
  			if (overallCount == 0 || isInOrder) {
  				latestCompletedCheckpoint = checkpoint;
  
  				// Clear cached stats
  				operatorStatsCache.clear();
  
  				// Update the stats per sub task
  				subTaskStats = statsForSubTasks;
  			}
  
  			long checkpointId = checkpoint.getCheckpointID();
  			long checkpointTriggerTimestamp = checkpoint.getTimestamp();
  			long checkpointDuration = checkpoint.getDuration();
  
  			overallCount++;
  
  			// Duration stats
  			if (checkpointDuration > overallMaxDuration) {
  				overallMaxDuration = checkpointDuration;
  			}
  
  			if (checkpointDuration < overallMinDuration) {
  				overallMinDuration = checkpointDuration;
  			}
  
  			overallTotalDuration += checkpointDuration;
  
  			// State size stats
  			if (overallStateSize < overallMinStateSize) {
  				overallMinStateSize = overallStateSize;
  			}
  
  			if (overallStateSize > overallMaxStateSize) {
  				overallMaxStateSize = overallStateSize;
  			}
  
  			this.overallTotalStateSize += overallStateSize;
  
  			// Recent history
  			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();
  
  					// Only remove it if it the new checkpoint is not too old
  					if (size == historySize) {
  						if (checkpointId > history.get(0).getCheckpointId()) {
  							history.remove(0);
  						}
  					}
  
  					int pos = 0;
  
  					// Find position
  					for (int i = 0; i < size; i++) {
  						pos = i;
  
  						if (checkpointId < history.get(i).getCheckpointId()) {
  							break;
  						}
  					}
  
  					history.add(pos, stats);
  				}
  			}
  		}
  	}
focal_method.class_method_signature: SimpleCheckpointStatsTracker.onCompletedCheckpoint(CompletedCheckpoint checkpoint)
focal_method.full_signature: @Override public void onCompletedCheckpoint(CompletedCheckpoint checkpoint)
focal_method.identifier: onCompletedCheckpoint
focal_method.invocations: isEmpty getStates getStateSize getOperatorId get get put getSubtask getDuration getStateSize getCheckpointID getCheckpointID clear getCheckpointID getTimestamp getDuration size remove add size getCheckpointId get remove getCheckpointId get add
focal_method.modifiers: @Override public
focal_method.parameters: (CompletedCheckpoint checkpoint)
focal_method.return: void
focal_method.signature: void onCompletedCheckpoint(CompletedCheckpoint checkpoint)