while (iterator.hasNext()) {
                PlanDetails details = iterator.next();
                WorkSource.State state = details.source.executionState();
                if (state == WorkSource.State.NoMoreWorkToStart) {
                    if (details.source.allExecutionComplete()) {
                        iterator.remove();
                    }

        }
    }

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

            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()
                if (selection.noMoreWorkToStart) {
                    break
                }