}

    public boolean isRequired() {
        return state == ExecutionState.SHOULD_RUN;
    }

    public boolean isDoNotIncludeInPlan() {
        return filtered || state == ExecutionState.NOT_SCHEDULED || isCannotRunInAnyPlan();
    }

    public boolean isCannotRunInAnyPlan() {
        return state == ExecutionState.EXECUTED || state == ExecutionState.FAILED_DEPENDENCY;
    }

    /**

            for (WorkerState worker : workers) {
                ExecutionState currentState = worker.state.get();
                if (currentState == ExecutionState.Running) {
                    return null;
                } else if (currentState == ExecutionState.Waiting) {
                    waitingWorkers++;
                } else if (currentState == ExecutionState.Stopped) {
                    stoppedWorkers++;
                }

    {
      tensorflow::mutex_lock l(execution_mutex_);
      while (execution_state_ == ExecutionState::kIdle ||
             execution_state_ == ExecutionState::kHasResult) {
        start_execute_.wait(l);
      }
      if (execution_state_ == ExecutionState::kShuttingDown) {
        return;
      } else if (execution_state_ == ExecutionState::kReadyToExecute) {
        // op_outputs_ may have been std::moved

        if (!(resourceLock instanceof WorkerLeaseRegistry.WorkerLease) && !readyNodes.isEmpty()) {
            maybeNodesSelectable = true;
        }
    }

    @Override
    public State executionState() {
        lockCoordinator.assertHasStateLock();
        if (waitingToStartNodes.isEmpty()) {
            return State.NoMoreWorkToStart;
        } else if (!readyNodes.isEmpty() && maybeNodesSelectable) {

        return tasks
    }

    List<Node> getExecutedNodes() {
        def nodes = []
        coordinator.withStateLock {
            while (finalizedPlan.executionState() != WorkSource.State.NoMoreWorkToStart) {
                assert finalizedPlan.executionState() == WorkSource.State.MaybeWorkReadyToStart // There should always be a node ready to start when executing sequentially
                def selection = finalizedPlan.selectNext()

    List<Deployment> deployments = []
    def continuousExecutionGate = new DefaultContinuousExecutionGate()
    def deploymentRegistry = Stub(DeploymentRegistryInternal) {
        runningDeployments >> deployments
        executionGate >> continuousExecutionGate
    }
    def buildSessionContext = Mock(BuildSessionContext)
    def textOutputFactory = new TestStyledTextOutputFactory()
    def executorService = Executors.newCachedThreadPool()