JobGraph implements ExecutionPlan { 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; } JobGraph(Config config, ApplicationDescriptorImpl appDesc); @Override List<JobConfig> getJobConfigs(); @Override List<StreamSpec> getIntermediateStreams(); @Override String getPlanAsJson(); @Override ApplicationConfig getApplicationConfig();  }
@Test public void testReachable() { Set<JobNode> reachable1 = graph1.findReachable(); assertTrue(reachable1.size() == 8); Set<JobNode> reachable2 = graph2.findReachable(); assertTrue(reachable2.size() == 7); }