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 testTopologicalSort3() {
        //             --> op1 --
        //            /         \
        //  (source) -           +-> op2 -> op3
        //            \         /
        //             ---------

        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, ResultPartitionType.PIPELINED);
            op2.connectNewDataSetAsInput(
                    op1, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
            op2.connectNewDataSetAsInput(
                    source, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
            op3.connectNewDataSetAsInput(
                    op2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);

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