@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, ResultPartitionType.PIPELINED); v2.connectNewDataSetAsInput(v1, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED); v3.connectNewDataSetAsInput(v2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED); v4.connectNewDataSetAsInput(v3, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED); 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()); } }
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 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 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(String jobName);  JobGraph(JobID jobId, String jobName);  JobGraph(JobVertex... vertices);  JobGraph(String jobName, JobVertex... vertices);  JobGraph(JobID jobId, String jobName, JobVertex... vertices); }
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(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(); SerializedValue<ExecutionConfig> getSerializedExecutionConfig(); long getSessionTimeout(); void setSessionTimeout(long sessionTimeout); void setAllowQueuedScheduling(boolean allowQueuedScheduling); boolean getAllowQueuedScheduling(); void setScheduleMode(ScheduleMode scheduleMode); ScheduleMode getScheduleMode(); void setSavepointRestoreSettings(SavepointRestoreSettings settings); SavepointRestoreSettings getSavepointRestoreSettings(); void setExecutionConfig(ExecutionConfig executionConfig); void addVertex(JobVertex vertex); Iterable<JobVertex> getVertices(); JobVertex[] getVerticesAsArray(); int getNumberOfVertices(); void setSnapshotSettings(JobCheckpointingSettings settings); JobCheckpointingSettings getCheckpointingSettings(); boolean isCheckpointingEnabled(); JobVertex findVertexByID(JobVertexID id); void setClasspaths(List<URL> paths); List<URL> getClasspaths(); int getMaximumParallelism(); List<JobVertex> getVerticesSortedTopologicallyFromSources(); void addJar(Path jar); List<Path> getUserJars(); void addUserArtifact(String name, DistributedCache.DistributedCacheEntry file); Map<String, DistributedCache.DistributedCacheEntry> getUserArtifacts(); void addUserJarBlobKey(PermanentBlobKey key); boolean hasUsercodeJarFiles(); List<PermanentBlobKey> getUserJarBlobKeys(); @Override String toString(); void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey); void writeUserArtifactEntriesToConfiguration(); }
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(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(); SerializedValue<ExecutionConfig> getSerializedExecutionConfig(); long getSessionTimeout(); void setSessionTimeout(long sessionTimeout); void setAllowQueuedScheduling(boolean allowQueuedScheduling); boolean getAllowQueuedScheduling(); void setScheduleMode(ScheduleMode scheduleMode); ScheduleMode getScheduleMode(); void setSavepointRestoreSettings(SavepointRestoreSettings settings); SavepointRestoreSettings getSavepointRestoreSettings(); void setExecutionConfig(ExecutionConfig executionConfig); void addVertex(JobVertex vertex); Iterable<JobVertex> getVertices(); JobVertex[] getVerticesAsArray(); int getNumberOfVertices(); void setSnapshotSettings(JobCheckpointingSettings settings); JobCheckpointingSettings getCheckpointingSettings(); boolean isCheckpointingEnabled(); JobVertex findVertexByID(JobVertexID id); void setClasspaths(List<URL> paths); List<URL> getClasspaths(); int getMaximumParallelism(); List<JobVertex> getVerticesSortedTopologicallyFromSources(); void addJar(Path jar); List<Path> getUserJars(); void addUserArtifact(String name, DistributedCache.DistributedCacheEntry file); Map<String, DistributedCache.DistributedCacheEntry> getUserArtifacts(); void addUserJarBlobKey(PermanentBlobKey key); boolean hasUsercodeJarFiles(); List<PermanentBlobKey> getUserJarBlobKeys(); @Override String toString(); void setUserArtifactBlobKey(String entryName, PermanentBlobKey blobKey); void writeUserArtifactEntriesToConfiguration();  }