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 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, ResultPartitionType.PIPELINED);
            root.connectNewDataSetAsInput(
                    source2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
            root.connectNewDataSetAsInput(
                    l2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);

            l2.connectNewDataSetAsInput(
                    l11, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
            l2.connectNewDataSetAsInput(
                    l12, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);

            l11.connectNewDataSetAsInput(
                    source1, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);

            l12.connectNewDataSetAsInput(
                    source1, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
            l12.connectNewDataSetAsInput(
                    source2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);

            l13.connectNewDataSetAsInput(
                    source2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);

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