flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
/**
	 * Verifies that all operator states can be mapped to an execution job vertex.
	 *
	 * @param allowNonRestoredState if false an exception will be thrown if a state could not be mapped
	 * @param operatorStates operator states to map
	 * @param tasks task to map to
	 */
Verifies that all operator states can be mapped to an execution job vertex.
@param allowNonRestoredState if false an exception will be thrown if a state could not be mapped
@param operatorStates operator states to map
@param tasks task to map to
Verifies that all operator states can be mapped to an execution job vertex . @param allowNonRestoredState if false an exception will be thrown if a state could not be mapped @param operatorStates operator states to map @param tasks task to map to
private static void checkStateMappingCompleteness(
			boolean allowNonRestoredState,
			Map<OperatorID, OperatorState> operatorStates,
			Map<JobVertexID, ExecutionJobVertex> tasks) {
		Set<OperatorID> allOperatorIDs = new HashSet<>();
		for (ExecutionJobVertex executionJobVertex : tasks.values()) {
			allOperatorIDs.addAll(executionJobVertex.getOperatorIDs());
		}
		for (Map.Entry<OperatorID, OperatorState> operatorGroupStateEntry : operatorStates.entrySet()) {
			OperatorState operatorState = operatorGroupStateEntry.getValue();
			if (!allOperatorIDs.contains(operatorGroupStateEntry.getKey())) {
				if (allowNonRestoredState) {
					LOG.info("Skipped checkpoint state for operator {}.", operatorState.getOperatorID());
				} else {
					throw new IllegalStateException("There is no operator for the state " + operatorState.getOperatorID());
				}
			}
		}
	}
private static void checkStateMappingCompleteness ( boolean allowNonRestoredState , Map < OperatorID , OperatorState > operatorStates , Map < JobVertexID , ExecutionJobVertex > tasks ) { Set < OperatorID > allOperatorIDs = new HashSet < > ( ) ; for ( ExecutionJobVertex executionJobVertex : tasks . values ( ) ) { allOperatorIDs . addAll ( executionJobVertex . getOperatorIDs ( ) ) ; } for ( Map . Entry < OperatorID , OperatorState > operatorGroupStateEntry : operatorStates . entrySet ( ) ) { OperatorState operatorState = operatorGroupStateEntry . getValue ( ) ; if ( ! allOperatorIDs . contains ( operatorGroupStateEntry . getKey ( ) ) ) { if ( allowNonRestoredState ) { LOG . info ( "Skipped checkpoint state for operator {}." , operatorState . getOperatorID ( ) ) ; } else { throw new IllegalStateException ( "There is no operator for the state " + operatorState . getOperatorID ( ) ) ; } } } }
Verifies that all operator states can be mapped to an execution job vertex.
Verifies that all operator states can be mapped to an execution job vertex .
//----------------------------------------find operator for state---------------------------------------------