code: static Set<DagNode<JobExecutionPlan>> getNext(Dag<JobExecutionPlan> dag) {
      Set<DagNode<JobExecutionPlan>> nextNodesToExecute = new HashSet<>();
      LinkedList<DagNode<JobExecutionPlan>> nodesToExpand = Lists.newLinkedList(dag.getStartNodes());
      FailureOption failureOption = getFailureOption(dag);
  
      while (!nodesToExpand.isEmpty()) {
        DagNode<JobExecutionPlan> node = nodesToExpand.poll();
        ExecutionStatus executionStatus = getExecutionStatus(node);
        boolean addFlag = true;
        if (executionStatus == ExecutionStatus.$UNKNOWN) {
          //Add a node to be executed next, only if all of its parent nodes are COMPLETE.
          List<DagNode<JobExecutionPlan>> parentNodes = dag.getParents(node);
          for (DagNode<JobExecutionPlan> parentNode : parentNodes) {
            if (getExecutionStatus(parentNode) != ExecutionStatus.COMPLETE) {
              addFlag = false;
              break;
            }
          }
          if (addFlag) {
            nextNodesToExecute.add(node);
          }
        } else if (executionStatus == ExecutionStatus.COMPLETE) {
          //Explore the children of COMPLETED node as next candidates for execution.
          nodesToExpand.addAll(dag.getChildren(node));
        } else if ((executionStatus == ExecutionStatus.FAILED) || (executionStatus == ExecutionStatus.CANCELLED)) {
          switch (failureOption) {
            case FINISH_RUNNING:
              return new HashSet<>();
            case FINISH_ALL_POSSIBLE:
            default:
              break;
          }
        }
      }
      return nextNodesToExecute;
    }
docs: Traverse the dag to determine the next set of nodes to be executed. It starts with the startNodes of the dag and
  identifies each node yet to be executed and for which each of its parent nodes is in the {@link ExecutionStatus#COMPLETE}
  state.