flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
/**
	 * Acknowledges the task with the given execution attempt id and the given subtask state.
	 *
	 * @param executionAttemptId of the acknowledged task
	 * @param operatorSubtaskStates of the acknowledged task
	 * @param metrics Checkpoint metrics for the stats
	 * @return TaskAcknowledgeResult of the operation
	 */
Acknowledges the task with the given execution attempt id and the given subtask state.
@param executionAttemptId of the acknowledged task
@param operatorSubtaskStates of the acknowledged task
@param metrics Checkpoint metrics for the stats
@return TaskAcknowledgeResult of the operation
Acknowledges the task with the given execution attempt id and the given subtask state . @param executionAttemptId of the acknowledged task @param operatorSubtaskStates of the acknowledged task @param metrics Checkpoint metrics for the stats @return TaskAcknowledgeResult of the operation
public TaskAcknowledgeResult acknowledgeTask(
			ExecutionAttemptID executionAttemptId,
			TaskStateSnapshot operatorSubtaskStates,
			CheckpointMetrics metrics) {
		synchronized (lock) {
			if (discarded) {
				return TaskAcknowledgeResult.DISCARDED;
			}
			final ExecutionVertex vertex = notYetAcknowledgedTasks.remove(executionAttemptId);
			if (vertex == null) {
				if (acknowledgedTasks.contains(executionAttemptId)) {
					return TaskAcknowledgeResult.DUPLICATE;
				} else {
					return TaskAcknowledgeResult.UNKNOWN;
				}
			} else {
				acknowledgedTasks.add(executionAttemptId);
			}
			List<OperatorID> operatorIDs = vertex.getJobVertex().getOperatorIDs();
			int subtaskIndex = vertex.getParallelSubtaskIndex();
			long ackTimestamp = System.currentTimeMillis();
			long stateSize = 0L;
			if (operatorSubtaskStates != null) {
				for (OperatorID operatorID : operatorIDs) {
					OperatorSubtaskState operatorSubtaskState =
						operatorSubtaskStates.getSubtaskStateByOperatorID(operatorID);
					if (operatorSubtaskState == null) {
						operatorSubtaskState = new OperatorSubtaskState();
					}
					OperatorState operatorState = operatorStates.get(operatorID);
					if (operatorState == null) {
						operatorState = new OperatorState(
							operatorID,
							vertex.getTotalNumberOfParallelSubtasks(),
							vertex.getMaxParallelism());
						operatorStates.put(operatorID, operatorState);
					}
					operatorState.putState(subtaskIndex, operatorSubtaskState);
					stateSize += operatorSubtaskState.getStateSize();
				}
			}
			++numAcknowledgedTasks;
			final PendingCheckpointStats statsCallback = this.statsCallback;
			if (statsCallback != null) {
				long alignmentDurationMillis = metrics.getAlignmentDurationNanos() / 1_000_000;
				SubtaskStateStats subtaskStateStats = new SubtaskStateStats(
					subtaskIndex,
					ackTimestamp,
					stateSize,
					metrics.getSyncDurationMillis(),
					metrics.getAsyncDurationMillis(),
					metrics.getBytesBufferedInAlignment(),
					alignmentDurationMillis);
				statsCallback.reportSubtaskStats(vertex.getJobvertexId(), subtaskStateStats);
			}
			return TaskAcknowledgeResult.SUCCESS;
		}
	}
public TaskAcknowledgeResult acknowledgeTask ( ExecutionAttemptID executionAttemptId , TaskStateSnapshot operatorSubtaskStates , CheckpointMetrics metrics ) { synchronized ( lock ) { if ( discarded ) { return TaskAcknowledgeResult . DISCARDED ; } final ExecutionVertex vertex = notYetAcknowledgedTasks . remove ( executionAttemptId ) ; if ( vertex == null ) { if ( acknowledgedTasks . contains ( executionAttemptId ) ) { return TaskAcknowledgeResult . DUPLICATE ; } else { return TaskAcknowledgeResult . UNKNOWN ; } } else { acknowledgedTasks . add ( executionAttemptId ) ; } List < OperatorID > operatorIDs = vertex . getJobVertex ( ) . getOperatorIDs ( ) ; int subtaskIndex = vertex . getParallelSubtaskIndex ( ) ; long ackTimestamp = System . currentTimeMillis ( ) ; long stateSize = 0L ; if ( operatorSubtaskStates != null ) { for ( OperatorID operatorID : operatorIDs ) { OperatorSubtaskState operatorSubtaskState = operatorSubtaskStates . getSubtaskStateByOperatorID ( operatorID ) ; if ( operatorSubtaskState == null ) { operatorSubtaskState = new OperatorSubtaskState ( ) ; } OperatorState operatorState = operatorStates . get ( operatorID ) ; if ( operatorState == null ) { operatorState = new OperatorState ( operatorID , vertex . getTotalNumberOfParallelSubtasks ( ) , vertex . getMaxParallelism ( ) ) ; operatorStates . put ( operatorID , operatorState ) ; } operatorState . putState ( subtaskIndex , operatorSubtaskState ) ; stateSize += operatorSubtaskState . getStateSize ( ) ; } } ++ numAcknowledgedTasks ; final PendingCheckpointStats statsCallback = this . statsCallback ; if ( statsCallback != null ) { long alignmentDurationMillis = metrics . getAlignmentDurationNanos ( ) / 1_000_000 ; SubtaskStateStats subtaskStateStats = new SubtaskStateStats ( subtaskIndex , ackTimestamp , stateSize , metrics . getSyncDurationMillis ( ) , metrics . getAsyncDurationMillis ( ) , metrics . getBytesBufferedInAlignment ( ) , alignmentDurationMillis ) ; statsCallback . reportSubtaskStats ( vertex . getJobvertexId ( ) , subtaskStateStats ) ; } return TaskAcknowledgeResult . SUCCESS ; } }
Acknowledges the task with the given execution attempt id and the given subtask state.
Acknowledges the task with the given execution attempt id and the given subtask state .
// if no real operatorSubtaskState was reported, we insert an empty state // publish the checkpoint statistics // to prevent null-pointers from concurrent modification, copy reference onto stack // Do this in millis because the web frontend works with them