class StateAssignmentOperation {

	public void assignStates() {
		Map<OperatorID, OperatorState> localOperators = new HashMap<>(operatorStates);

		checkStateMappingCompleteness(allowNonRestoredState, operatorStates, tasks);

		for (ExecutionJobVertex executionJobVertex : this.tasks) {

			// find the states of all operators belonging to this task
			List<OperatorIDPair> operatorIDPairs = executionJobVertex.getOperatorIDs();
			List<OperatorState> operatorStates = new ArrayList<>(operatorIDPairs.size());
			boolean statelessSubTasks = true;
			for (OperatorIDPair operatorIDPair : operatorIDPairs) {
				OperatorID operatorID = operatorIDPair.getUserDefinedOperatorID().orElse(operatorIDPair.getGeneratedOperatorID());

				OperatorState operatorState = localOperators.remove(operatorID);
				if (operatorState == null) {
					operatorState = new OperatorState(
						operatorID,
						executionJobVertex.getParallelism(),
						executionJobVertex.getMaxParallelism());
				} else if (operatorState.getNumberCollectedStates() > 0) {
					statelessSubTasks = false;
				}
				operatorStates.add(operatorState);
			}
			if (!statelessSubTasks) { // skip tasks where no operator has any state
				assignAttemptState(executionJobVertex, operatorStates);
			}
		}

	}

	public  StateAssignmentOperation(
		long restoreCheckpointId,
		Set<ExecutionJobVertex> tasks,
		Map<OperatorID, OperatorState> operatorStates,
		boolean allowNonRestoredState);

}