@Override
    public void handleGlobalFailure(final Throwable error) {
        final long timestamp = System.currentTimeMillis();
        setGlobalFailureCause(error, timestamp);

        log.info("Trying to recover from a global failure.", error);
        final FailureHandlingResult failureHandlingResult =
                executionFailureHandler.getGlobalFailureHandlingResult(error, timestamp);
        maybeRestartTasks(failureHandlingResult);
    }
@Test
    void testExceptionHistoryWithGlobalFailOver() {
        final JobGraph jobGraph = singleNonParallelJobVertexJobGraph();
        final DefaultScheduler scheduler = createSchedulerAndStartScheduling(jobGraph);

        final ExecutionAttemptID attemptId =
                Iterables.getOnlyElement(
                                scheduler
                                        .requestJob()
                                        .getArchivedExecutionGraph()
                                        .getAllExecutionVertices())
                        .getCurrentExecutionAttempt()
                        .getAttemptId();

        final Exception expectedException = new Exception("Expected exception");
        scheduler.handleGlobalFailure(expectedException);

        // we have to cancel the task and trigger the restart to have the exception history
        // populated
        scheduler.updateTaskExecutionState(
                new TaskExecutionState(attemptId, ExecutionState.CANCELED, expectedException));
        taskRestartExecutor.triggerScheduledTasks();

        final Iterable<RootExceptionHistoryEntry> actualExceptionHistory =
                scheduler.getExceptionHistory();

        assertThat(actualExceptionHistory).hasSize(1);

        final RootExceptionHistoryEntry failure = actualExceptionHistory.iterator().next();

        assertThat(
                        ExceptionHistoryEntryTestingUtils.matchesGlobalFailure(
                                failure,
                                expectedException,
                                scheduler.getExecutionGraph().getFailureInfo().getTimestamp()))
                .isTrue();
        assertThat(failure.getConcurrentExceptions()).isEmpty();
    }