@Override
    public void startScheduling() {
        final Set<SchedulingPipelinedRegion> sourceRegions =
                IterableUtils.toStream(schedulingTopology.getAllPipelinedRegions())
                        .filter(this::isSourceRegion)
                        .collect(Collectors.toSet());
        maybeScheduleRegions(sourceRegions);
    }
@Test
    void testSchedulingRegionWithInnerNonPipelinedEdge() throws Exception {
        final JobVertex v1 = createJobVertex("v1", 1);
        final JobVertex v2 = createJobVertex("v2", 1);
        final JobVertex v3 = createJobVertex("v3", 1);
        final JobVertex v4 = createJobVertex("v4", 1);
        final JobVertex v5 = createJobVertex("v5", 1);

        v2.connectNewDataSetAsInput(
                v1, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
        v3.connectNewDataSetAsInput(
                v2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
        v4.connectNewDataSetAsInput(
                v2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
        v5.connectNewDataSetAsInput(
                v2, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
        v3.connectNewDataSetAsInput(
                v1, DistributionPattern.POINTWISE, ResultPartitionType.HYBRID_FULL);
        v4.connectNewDataSetAsInput(
                v1, DistributionPattern.POINTWISE, ResultPartitionType.HYBRID_SELECTIVE);
        v4.connectNewDataSetAsInput(
                v1, DistributionPattern.POINTWISE, ResultPartitionType.BLOCKING);

        final List<JobVertex> ordered = new ArrayList<>(Arrays.asList(v1, v2, v3, v4, v5));
        final JobGraph jobGraph =
                JobGraphBuilder.newBatchJobGraphBuilder().addJobVertices(ordered).build();
        final ExecutionGraph executionGraph =
                TestingDefaultExecutionGraphBuilder.newBuilder()
                        .setJobGraph(jobGraph)
                        .build(EXECUTOR_RESOURCE.getExecutor());

        final SchedulingTopology schedulingTopology = executionGraph.getSchedulingTopology();

        startScheduling(schedulingTopology);

        assertThat(testingSchedulerOperation.getScheduledVertices()).hasSize(1);
        List<ExecutionVertexID> executionVertexIds =
                testingSchedulerOperation.getScheduledVertices().get(0);
        assertThat(executionVertexIds).hasSize(5);
    }