@Override
    public void startScheduling() {
        checkIdleSlotTimeout();
        state.as(Created.class)
                .orElseThrow(
                        () ->
                                new IllegalStateException(
                                        "Can only start scheduling when being in Created state."))
                .startScheduling();
    }
@Test
    void testNumRestartsMetric() throws Exception {
        final CompletableFuture<Gauge<Long>> numRestartsMetricFuture = new CompletableFuture<>();
        final MetricRegistry metricRegistry =
                TestingMetricRegistry.builder()
                        .setRegisterConsumer(
                                (metric, name, group) -> {
                                    if (MetricNames.NUM_RESTARTS.equals(name)) {
                                        numRestartsMetricFuture.complete((Gauge<Long>) metric);
                                    }
                                })
                        .build();

        final JobGraph jobGraph = createJobGraph();

        final DefaultDeclarativeSlotPool declarativeSlotPool =
                new DefaultDeclarativeSlotPool(
                        jobGraph.getJobID(),
                        new DefaultAllocatedSlotPool(),
                        ignored -> {},
                        Duration.ofMinutes(10),
                        Duration.ofMinutes(10),
                        Duration.ZERO,
                        mainThreadExecutor);

        final Configuration configuration = createConfigurationWithNoTimeouts();
        configuration.set(JobManagerOptions.MIN_PARALLELISM_INCREASE, 1);

        final AdaptiveScheduler scheduler =
                new AdaptiveSchedulerBuilder(
                                jobGraph,
                                singleThreadMainThreadExecutor,
                                EXECUTOR_RESOURCE.getExecutor())
                        .setJobMasterConfiguration(configuration)
                        .setJobManagerJobMetricGroup(
                                JobManagerMetricGroup.createJobManagerMetricGroup(
                                                metricRegistry, "localhost")
                                        .addJob(new JobID(), "jobName"))
                        .setDeclarativeSlotPool(declarativeSlotPool)
                        .build();

        final Gauge<Long> numRestartsMetric = numRestartsMetricFuture.get();

        final SubmissionBufferingTaskManagerGateway taskManagerGateway =
                new SubmissionBufferingTaskManagerGateway(1 + PARALLELISM);

        taskManagerGateway.setCancelConsumer(createCancelConsumer(scheduler));

        singleThreadMainThreadExecutor.execute(
                () -> {
                    scheduler.startScheduling();

                    declarativeSlotPool.offerSlots(
                            createSlotOffersForResourceRequirements(
                                    ResourceCounter.withResource(ResourceProfile.UNKNOWN, 1)),
                            new LocalTaskManagerLocation(),
                            taskManagerGateway,
                            System.currentTimeMillis());
                });

        // wait for the first task submission
        taskManagerGateway.waitForSubmissions(1);

        assertThat(numRestartsMetric.getValue()).isEqualTo(0L);

        singleThreadMainThreadExecutor.execute(
                () -> {
                    // offer more slots, which will cause a restart in order to scale up
                    offerSlots(
                            declarativeSlotPool,
                            createSlotOffersForResourceRequirements(
                                    ResourceCounter.withResource(
                                            ResourceProfile.UNKNOWN, PARALLELISM)),
                            taskManagerGateway);
                });

        // wait for the second task submissions
        taskManagerGateway.waitForSubmissions(PARALLELISM);

        assertThat(numRestartsMetric.getValue()).isEqualTo(1L);
    }