code: 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();
  			//----------------------------------------find operator for state---------------------------------------------
  
  			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());
  				}
  			}
  		}
  	}
docs: 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