Set<JobNode> findReachable() {
    Queue<JobNode> queue = new ArrayDeque<>();
    Set<JobNode> visited = new HashSet<>();

    inputStreams.forEach(input -> {
      List<JobNode> next = input.getTargetNodes();
      queue.addAll(next);
      visited.addAll(next);
    });

    while (!queue.isEmpty()) {
      JobNode node = queue.poll();
      node.getOutEdges().values().stream().flatMap(edge -> edge.getTargetNodes().stream()).forEach(target -> {
        if (!visited.contains(target)) {
          visited.add(target);
          queue.offer(target);
        }
      });
    }

    return visited;
  }
Find the reachable set of nodes using BFS.
@return reachable set of {@link JobNode}
samza-core/src/main/java/org/apache/samza/execution/JobGraph.java