@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 testStartScheduling() {
        VertexwiseSchedulingStrategy schedulingStrategy =
                createSchedulingStrategy(testingSchedulingTopology);

        final List<List<TestingSchedulingExecutionVertex>> expectedScheduledVertices =
                new ArrayList<>();
        expectedScheduledVertices.add(Collections.singletonList(source.get(0)));
        expectedScheduledVertices.add(Collections.singletonList(source.get(1)));

        inputConsumableDecider.addSourceVertices(new HashSet<>(source));

        schedulingStrategy.startScheduling();
        assertLatestScheduledVerticesAreEqualTo(
                expectedScheduledVertices, testingSchedulerOperation);
    }