@Override
    public void handleGlobalFailure(Throwable cause) {
        final FailureEnricher.Context ctx =
                DefaultFailureEnricherContext.forGlobalFailure(
                        jobInfo, jobManagerJobMetricGroup, ioExecutor, userCodeClassLoader);
        final CompletableFuture<Map<String, String>> failureLabels =
                FailureEnricherUtils.labelFailure(
                        cause, ctx, getMainThreadExecutor(), failureEnrichers);
        state.handleGlobalFailure(cause, failureLabels);
    }
@Test
    void testExceptionHistoryWithGlobalFailureLabels() throws Exception {
        final Exception expectedException = new Exception("Global Exception to label");
        BiConsumer<AdaptiveScheduler, List<ExecutionAttemptID>> testLogic =
                (scheduler, attemptIds) -> scheduler.handleGlobalFailure(expectedException);

        final TestingFailureEnricher failureEnricher = new TestingFailureEnricher();
        final Iterable<RootExceptionHistoryEntry> actualExceptionHistory =
                new ExceptionHistoryTester(singleThreadMainThreadExecutor)
                        .withTestLogic(testLogic)
                        .withFailureEnrichers(Collections.singletonList(failureEnricher))
                        .run();

        assertThat(actualExceptionHistory).hasSize(1);

        final RootExceptionHistoryEntry failure = actualExceptionHistory.iterator().next();
        assertThat(failure.getTaskManagerLocation()).isNull();
        assertThat(failure.getFailingTaskName()).isNull();
        assertThat(failureEnricher.getSeenThrowables()).containsExactly(expectedException);
        assertThat(failure.getFailureLabels()).isEqualTo(failureEnricher.getFailureLabels());

        assertThat(failure.getException().deserializeError(classLoader))
                .isEqualTo(expectedException);
    }