@Override
    public void startScheduling() {
        checkIdleSlotTimeout();
        state.as(Created.class)
                .orElseThrow(
                        () ->
                                new IllegalStateException(
                                        "Can only start scheduling when being in Created state."))
                .startScheduling();
    }
@Test
    void testStatusMetrics() throws Exception {
        final CompletableFuture<UpTimeGauge> upTimeMetricFuture = new CompletableFuture<>();
        final CompletableFuture<DownTimeGauge> downTimeMetricFuture = new CompletableFuture<>();
        // restartingTime acts as a stand-in for generic status time metrics
        final CompletableFuture<Gauge<Long>> restartTimeMetricFuture = new CompletableFuture<>();
        final MetricRegistry metricRegistry =
                TestingMetricRegistry.builder()
                        .setRegisterConsumer(
                                (metric, name, group) -> {
                                    switch (name) {
                                        case UpTimeGauge.METRIC_NAME:
                                            upTimeMetricFuture.complete((UpTimeGauge) metric);
                                            break;
                                        case DownTimeGauge.METRIC_NAME:
                                            downTimeMetricFuture.complete((DownTimeGauge) metric);
                                            break;
                                        case "restartingTimeTotal":
                                            restartTimeMetricFuture.complete((Gauge<Long>) metric);
                                            break;
                                    }
                                })
                        .build();

        final JobGraph jobGraph = createJobGraph();

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

        final Configuration configuration = createConfigurationWithNoTimeouts();
        configuration.set(JobManagerOptions.MIN_PARALLELISM_INCREASE, 1);
        configuration.set(JobManagerOptions.RESOURCE_WAIT_TIMEOUT, Duration.ofMillis(10L));
        configuration.set(
                MetricOptions.JOB_STATUS_METRICS,
                Arrays.asList(MetricOptions.JobStatusMetrics.TOTAL_TIME));

        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 UpTimeGauge upTimeGauge = upTimeMetricFuture.get();
        final DownTimeGauge downTimeGauge = downTimeMetricFuture.get();
        final Gauge<Long> restartTimeGauge = restartTimeMetricFuture.get();

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

        taskManagerGateway.setCancelConsumer(createCancelConsumer(scheduler));

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

                    offerSlots(
                            declarativeSlotPool,
                            createSlotOffersForResourceRequirements(
                                    ResourceCounter.withResource(ResourceProfile.UNKNOWN, 1)),
                            taskManagerGateway);
                });

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

        CommonTestUtils.waitUntilCondition(() -> upTimeGauge.getValue() > 0L);
        assertThat(downTimeGauge.getValue()).isEqualTo(0L);
        assertThat(restartTimeGauge.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, 1)),
                            taskManagerGateway);
                });

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

        CommonTestUtils.waitUntilCondition(() -> upTimeGauge.getValue() > 0L);
        assertThat(downTimeGauge.getValue()).isEqualTo(0L);
        // can be zero if the restart is very quick
        assertThat(restartTimeGauge.getValue()).isGreaterThanOrEqualTo(0L);
    }