@Override
    public void startScheduling() {
        final Set<SchedulingPipelinedRegion> sourceRegions =
                IterableUtils.toStream(schedulingTopology.getAllPipelinedRegions())
                        .filter(this::isSourceRegion)
                        .collect(Collectors.toSet());
        maybeScheduleRegions(sourceRegions);
    }
@Test
    void testStartScheduling() {
        startScheduling(testingSchedulingTopology);

        final List<List<TestingSchedulingExecutionVertex>> expectedScheduledVertices =
                new ArrayList<>();
        expectedScheduledVertices.add(Arrays.asList(source.get(0), map1.get(0)));
        expectedScheduledVertices.add(Arrays.asList(source.get(1), map1.get(1)));
        expectedScheduledVertices.add(Arrays.asList(map2.get(0)));
        expectedScheduledVertices.add(Arrays.asList(map2.get(1)));
        expectedScheduledVertices.add(Arrays.asList(map3.get(0)));
        expectedScheduledVertices.add(Arrays.asList(map3.get(1)));
        assertLatestScheduledVerticesAreEqualTo(
                expectedScheduledVertices, testingSchedulerOperation);
    }