JobGraph implements ExecutionPlan { List<JobNode> topologicalSort() { Collection<JobNode> pnodes = nodes.values(); if (pnodes.size() == 1) { return new ArrayList<>(pnodes); } Queue<JobNode> q = new ArrayDeque<>(); Map<String, Long> indegree = new HashMap<>(); Set<JobNode> visited = new HashSet<>(); pnodes.forEach(node -> { String nid = node.getJobNameAndId(); long degree = node.getInEdges().values().stream().filter(e -> !inputStreams.contains(e)).count(); indegree.put(nid, degree); if (degree == 0L) { q.add(node); visited.add(node); } }); List<JobNode> sortedNodes = new ArrayList<>(); Set<JobNode> reachable = new HashSet<>(); while (sortedNodes.size() < pnodes.size()) { while (!q.isEmpty()) { JobNode node = q.poll(); sortedNodes.add(node); node.getOutEdges().values().stream().flatMap(edge -> edge.getTargetNodes().stream()).forEach(n -> { String nid = n.getJobNameAndId(); Long degree = indegree.get(nid) - 1; indegree.put(nid, degree); if (degree == 0L && !visited.contains(n)) { q.add(n); visited.add(n); } reachable.add(n); }); } if (sortedNodes.size() < pnodes.size()) { reachable.removeAll(sortedNodes); if (!reachable.isEmpty()) { long min = Long.MAX_VALUE; JobNode minNode = null; for (JobNode node : reachable) { Long degree = indegree.get(node.getJobNameAndId()); if (degree < min) { min = degree; minNode = node; } } q.add(minNode); visited.add(minNode); } else { JobNode nextNode = inputStreams.stream().flatMap(input -> input.getTargetNodes().stream()) .filter(node -> !visited.contains(node)) .findAny().get(); q.add(nextNode); visited.add(nextNode); } } } return sortedNodes; } JobGraph(Config config, ApplicationDescriptorImpl appDesc); @Override List<JobConfig> getJobConfigs(); @Override List<StreamSpec> getIntermediateStreams(); @Override String getPlanAsJson(); @Override ApplicationConfig getApplicationConfig();  }
@Test public void testTopologicalSort() { List<JobNode> sortedNodes1 = graph1.topologicalSort(); Map<String, Integer> idxMap1 = new HashMap<>(); for (int i = 0; i < sortedNodes1.size(); i++) { idxMap1.put(sortedNodes1.get(i).getJobName(), i); } assertTrue(idxMap1.size() == 8); assertTrue(idxMap1.get("11") > idxMap1.get("5")); assertTrue(idxMap1.get("11") > idxMap1.get("7")); assertTrue(idxMap1.get("8") > idxMap1.get("7")); assertTrue(idxMap1.get("8") > idxMap1.get("3")); assertTrue(idxMap1.get("2") > idxMap1.get("11")); assertTrue(idxMap1.get("9") > idxMap1.get("8")); assertTrue(idxMap1.get("9") > idxMap1.get("11")); assertTrue(idxMap1.get("10") > idxMap1.get("11")); assertTrue(idxMap1.get("10") > idxMap1.get("3")); List<JobNode> sortedNodes2 = graph2.topologicalSort(); Map<String, Integer> idxMap2 = new HashMap<>(); for (int i = 0; i < sortedNodes2.size(); i++) { idxMap2.put(sortedNodes2.get(i).getJobName(), i); } assertTrue(idxMap2.size() == 7); assertTrue(idxMap2.get("2") > idxMap2.get("1")); assertTrue(idxMap2.get("3") > idxMap2.get("1")); assertTrue(idxMap2.get("4") > idxMap2.get("1")); assertTrue(idxMap2.get("6") > idxMap2.get("1")); assertTrue(idxMap2.get("5") > idxMap2.get("4")); assertTrue(idxMap2.get("7") > idxMap2.get("5")); List<JobNode> sortedNodes3 = graph3.topologicalSort(); assertTrue(sortedNodes3.size() == 2); assertEquals(sortedNodes3.get(0).getJobName(), "1"); assertEquals(sortedNodes3.get(1).getJobName(), "2"); List<JobNode> sortedNodes4 = graph4.topologicalSort(); assertTrue(sortedNodes4.size() == 1); assertEquals(sortedNodes4.get(0).getJobName(), "1"); }