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 no real operatorSubtaskState was reported, we insert an empty state 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 ; // publish the checkpoint statistics // to prevent null-pointers from concurrent modification, copy reference onto stack final PendingCheckpointStats statsCallback = this . statsCallback ; if ( statsCallback != null ) { // Do this in millis because the web frontend works with them 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 .