@Override
    public void startScheduling() {
        Set<ExecutionVertexID> sourceVertices =
                IterableUtils.toStream(schedulingTopology.getVertices())
                        .filter(vertex -> vertex.getConsumedPartitionGroups().isEmpty())
                        .map(SchedulingExecutionVertex::getId)
                        .collect(Collectors.toSet());

        maybeScheduleVertices(sourceVertices);
    }
@Test
    void testScheduleDownstreamOfHybridEdge() {
        final TestingSchedulingTopology topology = new TestingSchedulingTopology();

        final List<TestingSchedulingExecutionVertex> producers =
                topology.addExecutionVertices().withParallelism(2).finish();

        final List<TestingSchedulingExecutionVertex> consumers =
                topology.addExecutionVertices().withParallelism(2).finish();

        // add consumers to scheduling strategy.
        topology.connectAllToAll(producers, consumers)
                .withResultPartitionType(ResultPartitionType.HYBRID_FULL)
                .finish();

        final VertexwiseSchedulingStrategy schedulingStrategy = createSchedulingStrategy(topology);
        inputConsumableDecider.addSourceVertices(new HashSet<>(producers));

        inputConsumableDecider.setInputConsumable(consumers.get(0));
        inputConsumableDecider.setInputConsumable(consumers.get(1));

        schedulingStrategy.startScheduling();

        // consumers are properly scheduled indicates that the consuming relationship and
        // correlation are successfully built
        assertLatestScheduledVerticesAreEqualTo(
                Arrays.asList(
                        Collections.singletonList(producers.get(0)),
                        Collections.singletonList(producers.get(1)),
                        Collections.singletonList(consumers.get(0)),
                        Collections.singletonList(consumers.get(1))),
                testingSchedulerOperation);
    }