public List<JobVertex> getVerticesSortedTopologicallyFromSources()
            throws InvalidProgramException {
        // early out on empty lists
        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());

        // start by finding the vertices with no input edges
        // and the ones with disconnected inputs (that refer to some standalone data set)
        {
            Iterator<JobVertex> iter = remaining.iterator();
            while (iter.hasNext()) {
                JobVertex vertex = iter.next();

                if (vertex.hasNoConnectedInputs()) {
                    sorted.add(vertex);
                    iter.remove();
                }
            }
        }

        int startNodePos = 0;

        // traverse from the nodes that were added until we found all elements
        while (!remaining.isEmpty()) {

            // first check if we have more candidates to start traversing from. if not, then the
            // graph is cyclic, which is not permitted
            if (startNodePos >= sorted.size()) {
                throw new InvalidProgramException("The job graph is cyclic.");
            }

            JobVertex current = sorted.get(startNodePos++);
            addNodesThatHaveNoNewPredecessors(current, sorted, remaining);
        }

        return sorted;
    }
@Test
    public void testTopologicalSort1() {
        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, ResultPartitionType.PIPELINED);
        target2.connectNewDataSetAsInput(
                source1, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
        target2.connectNewDataSetAsInput(
                intermediate2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
        intermediate2.connectNewDataSetAsInput(
                intermediate1, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
        intermediate1.connectNewDataSetAsInput(
                source2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);

        JobGraph graph =
                JobGraphTestUtils.streamingJobGraph(
                        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);
    }