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 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, ResultPartitionType.PIPELINED);
            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);
            target.connectNewDataSetAsInput(
                    v3, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);

            JobGraph jg = JobGraphTestUtils.streamingJobGraph(v1, v2, v3, v4, source, target);
            try {
                jg.getVerticesSortedTopologicallyFromSources();
                fail("Failed to raise error on topologically sorting cyclic graph.");
            } catch (InvalidProgramException e) {
                // that what we wanted
            }
        } catch (Exception e) {
            e.printStackTrace();
            fail(e.getMessage());
        }
    }