JobGraph implements Serializable { public List<JobVertex> getVerticesSortedTopologicallyFromSources() throws InvalidProgramException { if (this.taskVertices.isEmpty()) { return Collections.emptyList(); } List<JobVertex> sorted = new ArrayList<JobVertex>(this.taskVertices.size()); Set<JobVertex> remaining = new LinkedHashSet<JobVertex>(this.taskVertices.values()); { Iterator<JobVertex> iter = remaining.iterator(); while (iter.hasNext()) { JobVertex vertex = iter.next(); if (vertex.hasNoConnectedInputs()) { sorted.add(vertex); iter.remove(); } } } int startNodePos = 0; while (!remaining.isEmpty()) { if (startNodePos >= sorted.size()) { throw new InvalidProgramException("The job graph is cyclic."); } JobVertex current = sorted.get(startNodePos++); addNodesThatHaveNoNewPredecessors(current, sorted, remaining); } return sorted; } JobGraph();  JobGraph(String jobName);  JobGraph(JobID jobId, String jobName);  JobGraph(JobVertex... vertices);  JobGraph(String jobName, JobVertex... vertices);  JobGraph(JobID jobId, String jobName, JobVertex... vertices); JobID getJobID(); String getName(); Configuration getJobConfiguration(); void setRestartStrategyConfiguration(RestartStrategies.RestartStrategyConfiguration restartStrategyConfiguration); RestartStrategies.RestartStrategyConfiguration getRestartStrategyConfiguration(); long getSessionTimeout(); void setSessionTimeout(long sessionTimeout); void setAllowQueuedScheduling(boolean allowQueuedScheduling); boolean getAllowQueuedScheduling(); void setScheduleMode(ScheduleMode scheduleMode); ScheduleMode getScheduleMode(); void addVertex(JobVertex vertex); Iterable<JobVertex> getVertices(); JobVertex[] getVerticesAsArray(); int getNumberOfVertices(); void setSnapshotSettings(JobSnapshottingSettings settings); JobSnapshottingSettings getSnapshotSettings(); JobVertex findVertexByID(JobVertexID id); void setClasspaths(List<URL> paths); List<URL> getClasspaths(); void setSavepointPath(String savepointPath); List<JobVertex> getVerticesSortedTopologicallyFromSources(); void addJar(Path jar); void addBlob(BlobKey key); boolean hasUsercodeJarFiles(); List<BlobKey> getUserJarBlobKeys(); void uploadRequiredJarFiles(InetSocketAddress serverAddress); int getMaximumParallelism(); @Override String toString();  }
@Test public void testTopologicalSort1() { try { JobVertex source1 = new JobVertex("source1"); JobVertex source2 = new JobVertex("source2"); JobVertex target1 = new JobVertex("target1"); JobVertex target2 = new JobVertex("target2"); JobVertex intermediate1 = new JobVertex("intermediate1"); JobVertex intermediate2 = new JobVertex("intermediate2"); target1.connectNewDataSetAsInput(source1, DistributionPattern.POINTWISE); target2.connectNewDataSetAsInput(source1, DistributionPattern.POINTWISE); target2.connectNewDataSetAsInput(intermediate2, DistributionPattern.POINTWISE); intermediate2.connectNewDataSetAsInput(intermediate1, DistributionPattern.POINTWISE); intermediate1.connectNewDataSetAsInput(source2, DistributionPattern.POINTWISE); JobGraph graph = new JobGraph("TestGraph", source1, source2, intermediate1, intermediate2, target1, target2); List<JobVertex> sorted = graph.getVerticesSortedTopologicallyFromSources(); assertEquals(6, sorted.size()); assertBefore(source1, target1, sorted); assertBefore(source1, target2, sorted); assertBefore(source2, target2, sorted); assertBefore(source2, intermediate1, sorted); assertBefore(source2, intermediate2, sorted); assertBefore(intermediate1, target2, sorted); assertBefore(intermediate2, target2, sorted); } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }
@Test public void testTopologicalSort2() { try { JobVertex source1 = new JobVertex("source1"); JobVertex source2 = new JobVertex("source2"); JobVertex root = new JobVertex("root"); JobVertex l11 = new JobVertex("layer 1 - 1"); JobVertex l12 = new JobVertex("layer 1 - 2"); JobVertex l13 = new JobVertex("layer 1 - 3"); JobVertex l2 = new JobVertex("layer 2"); root.connectNewDataSetAsInput(l13, DistributionPattern.POINTWISE); root.connectNewDataSetAsInput(source2, DistributionPattern.POINTWISE); root.connectNewDataSetAsInput(l2, DistributionPattern.POINTWISE); l2.connectNewDataSetAsInput(l11, DistributionPattern.POINTWISE); l2.connectNewDataSetAsInput(l12, DistributionPattern.POINTWISE); l11.connectNewDataSetAsInput(source1, DistributionPattern.POINTWISE); l12.connectNewDataSetAsInput(source1, DistributionPattern.POINTWISE); l12.connectNewDataSetAsInput(source2, DistributionPattern.POINTWISE); l13.connectNewDataSetAsInput(source2, DistributionPattern.POINTWISE); JobGraph graph = new JobGraph("TestGraph", source1, source2, root, l11, l13, l12, l2); List<JobVertex> sorted = graph.getVerticesSortedTopologicallyFromSources(); assertEquals(7, sorted.size()); assertBefore(source1, root, sorted); assertBefore(source2, root, sorted); assertBefore(l11, root, sorted); assertBefore(l12, root, sorted); assertBefore(l13, root, sorted); assertBefore(l2, root, sorted); assertBefore(l11, l2, sorted); assertBefore(l12, l2, sorted); assertBefore(l2, root, sorted); assertBefore(source1, l2, sorted); assertBefore(source2, l2, sorted); assertBefore(source2, l13, sorted); } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }
@Test public void testTopologicalSort3() { try { JobVertex source = new JobVertex("source"); JobVertex op1 = new JobVertex("op4"); JobVertex op2 = new JobVertex("op2"); JobVertex op3 = new JobVertex("op3"); op1.connectNewDataSetAsInput(source, DistributionPattern.POINTWISE); op2.connectNewDataSetAsInput(op1, DistributionPattern.POINTWISE); op2.connectNewDataSetAsInput(source, DistributionPattern.POINTWISE); op3.connectNewDataSetAsInput(op2, DistributionPattern.POINTWISE); JobGraph graph = new JobGraph("TestGraph", source, op1, op2, op3); List<JobVertex> sorted = graph.getVerticesSortedTopologicallyFromSources(); assertEquals(4, sorted.size()); assertBefore(source, op1, sorted); assertBefore(source, op2, sorted); assertBefore(op1, op2, sorted); assertBefore(op2, op3, sorted); } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }
@Test public void testTopoSortCyclicGraphNoSources() { try { JobVertex v1 = new JobVertex("1"); JobVertex v2 = new JobVertex("2"); JobVertex v3 = new JobVertex("3"); JobVertex v4 = new JobVertex("4"); v1.connectNewDataSetAsInput(v4, DistributionPattern.POINTWISE); v2.connectNewDataSetAsInput(v1, DistributionPattern.POINTWISE); v3.connectNewDataSetAsInput(v2, DistributionPattern.POINTWISE); v4.connectNewDataSetAsInput(v3, DistributionPattern.POINTWISE); JobGraph jg = new JobGraph("Cyclic Graph", v1, v2, v3, v4); try { jg.getVerticesSortedTopologicallyFromSources(); fail("Failed to raise error on topologically sorting cyclic graph."); } catch (InvalidProgramException e) { } } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }
@Test public void testTopoSortCyclicGraphIntermediateCycle() { try{ JobVertex source = new JobVertex("source"); JobVertex v1 = new JobVertex("1"); JobVertex v2 = new JobVertex("2"); JobVertex v3 = new JobVertex("3"); JobVertex v4 = new JobVertex("4"); JobVertex target = new JobVertex("target"); v1.connectNewDataSetAsInput(source, DistributionPattern.POINTWISE); v1.connectNewDataSetAsInput(v4, DistributionPattern.POINTWISE); v2.connectNewDataSetAsInput(v1, DistributionPattern.POINTWISE); v3.connectNewDataSetAsInput(v2, DistributionPattern.POINTWISE); v4.connectNewDataSetAsInput(v3, DistributionPattern.POINTWISE); target.connectNewDataSetAsInput(v3, DistributionPattern.POINTWISE); JobGraph jg = new JobGraph("Cyclic Graph", v1, v2, v3, v4, source, target); try { jg.getVerticesSortedTopologicallyFromSources(); fail("Failed to raise error on topologically sorting cyclic graph."); } catch (InvalidProgramException e) { } } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }