@Override
    public void startScheduling() {
        checkIdleSlotTimeout();
        state.as(Created.class)
                .orElseThrow(
                        () ->
                                new IllegalStateException(
                                        "Can only start scheduling when being in Created state."))
                .startScheduling();
    }
@Test
    void testConsistentMaxParallelism() throws Exception {
        final int parallelism = 240;
        final int expectedMaxParallelism =
                KeyGroupRangeAssignment.computeDefaultMaxParallelism(parallelism);
        final JobVertex vertex = createNoOpVertex(parallelism);
        final JobGraph jobGraph = streamingJobGraph(vertex);

        final DefaultDeclarativeSlotPool declarativeSlotPool =
                createDeclarativeSlotPool(jobGraph.getJobID());

        final AdaptiveScheduler scheduler =
                new AdaptiveSchedulerBuilder(
                                jobGraph,
                                singleThreadMainThreadExecutor,
                                EXECUTOR_RESOURCE.getExecutor())
                        .setDeclarativeSlotPool(declarativeSlotPool)
                        .setJobMasterConfiguration(createConfigurationWithNoTimeouts())
                        .build();

        final SubmissionBufferingTaskManagerGateway taskManagerGateway =
                new SubmissionBufferingTaskManagerGateway(1 + parallelism);
        taskManagerGateway.setCancelConsumer(createCancelConsumer(scheduler));

        // offer just enough resources to run at the lowest possible parallelism
        singleThreadMainThreadExecutor.execute(
                () -> {
                    scheduler.startScheduling();
                    offerSlots(
                            declarativeSlotPool,
                            createSlotOffersForResourceRequirements(
                                    ResourceCounter.withResource(ResourceProfile.UNKNOWN, 1)),
                            taskManagerGateway);
                });

        // Wait for task to be submitted
        taskManagerGateway.waitForSubmissions(1);

        ArchivedExecutionGraph executionGraph =
                getArchivedExecutionGraphForRunningJob(scheduler).get();
        ArchivedExecutionJobVertex archivedVertex = executionGraph.getJobVertex(vertex.getID());

        // ensure that the parallelism was submitted based on what is available
        assertThat(archivedVertex.getParallelism()).isOne();
        // and that the max parallelism was submitted based on what was configured
        assertThat(archivedVertex.getMaxParallelism()).isEqualTo(expectedMaxParallelism);

        // offer the resources to run at full parallelism
        singleThreadMainThreadExecutor.execute(
                () -> {
                    offerSlots(
                            declarativeSlotPool,
                            createSlotOffersForResourceRequirements(
                                    ResourceCounter.withResource(
                                            ResourceProfile.UNKNOWN, parallelism)),
                            taskManagerGateway);
                });

        // wait for the job to be re-submitted
        taskManagerGateway.waitForSubmissions(parallelism);

        ArchivedExecutionGraph resubmittedExecutionGraph =
                getArchivedExecutionGraphForRunningJob(scheduler).get();
        ArchivedExecutionJobVertex resubmittedArchivedVertex =
                resubmittedExecutionGraph.getJobVertex(vertex.getID());

        // ensure that the parallelism was submitted based on what is available
        assertThat(resubmittedArchivedVertex.getParallelism()).isEqualTo(parallelism);
        // and that the max parallelism was submitted based on what was configured
        assertThat(resubmittedArchivedVertex.getMaxParallelism()).isEqualTo(expectedMaxParallelism);
    }