@Override
    public FailureResult howToHandleFailure(
            Throwable failure, CompletableFuture<Map<String, String>> failureLabels) {
        FailureResult failureResult = howToHandleFailure(failure);
        if (reportEventsAsSpans) {
            // TODO: replace with reporting as event once events are supported.
            // Add reporting as callback for when the failure labeling is completed.
            failureLabels.thenAcceptAsync(
                    (labels) -> jobFailureMetricReporter.reportJobFailure(failureResult, labels),
                    componentMainThreadExecutor);
        }
        return failureResult;
    }
@Test
    void testHowToHandleFailureAllowedByStrategy() throws Exception {
        final Configuration configuration = new Configuration();
        configuration.set(TraceOptions.REPORT_EVENTS_AS_SPANS, Boolean.TRUE);
        final List<Span> spanCollector = new ArrayList<>(1);
        final UnregisteredMetricGroups.UnregisteredJobManagerJobMetricGroup testMetricGroup =
                createTestMetricGroup(spanCollector);
        final TestRestartBackoffTimeStrategy restartBackoffTimeStrategy =
                new TestRestartBackoffTimeStrategy(true, 1234);

        final AdaptiveScheduler scheduler =
                new AdaptiveSchedulerBuilder(
                                createJobGraph(),
                                mainThreadExecutor,
                                EXECUTOR_RESOURCE.getExecutor())
                        .setRestartBackoffTimeStrategy(restartBackoffTimeStrategy)
                        .setJobMasterConfiguration(configuration)
                        .setJobManagerJobMetricGroup(testMetricGroup)
                        .build();

        final FailureResult failureResult =
                scheduler.howToHandleFailure(new Exception("test"), createFailureLabelsFuture());

        assertThat(failureResult.canRestart()).isTrue();
        assertThat(failureResult.getBackoffTime().toMillis())
                .isEqualTo(restartBackoffTimeStrategy.getBackoffTime());

        assertThat(spanCollector).isEmpty();
        mainThreadExecutor.trigger();
        checkMetrics(spanCollector, true);
    }