@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 handleGlobalFailureWithLocalFailure() {
        final JobGraph jobGraph = singleJobVertexJobGraph(2);
        final JobVertex onlyJobVertex = getOnlyJobVertex(jobGraph);
        enableCheckpointing(jobGraph);

        final DefaultScheduler scheduler = createSchedulerAndStartScheduling(jobGraph);

        final List<ExecutionAttemptID> attemptIds =
                StreamSupport.stream(
                                scheduler
                                        .requestJob()
                                        .getArchivedExecutionGraph()
                                        .getAllExecutionVertices()
                                        .spliterator(),
                                false)
                        .map(ArchivedExecutionVertex::getCurrentExecutionAttempt)
                        .map(ArchivedExecution::getAttemptId)
                        .collect(Collectors.toList());
        final ExecutionAttemptID localFailureAttemptId = attemptIds.get(0);
        scheduler.handleGlobalFailure(new Exception("global failure"));
        // the local failure shouldn't affect the global fail-over
        scheduler.updateTaskExecutionState(
                new TaskExecutionState(
                        localFailureAttemptId,
                        ExecutionState.FAILED,
                        new Exception("local failure")));

        for (ExecutionAttemptID attemptId : attemptIds) {
            scheduler.updateTaskExecutionState(
                    new TaskExecutionState(attemptId, ExecutionState.CANCELED));
        }

        taskRestartExecutor.triggerScheduledTasks();

        final ExecutionVertexID executionVertexId0 =
                new ExecutionVertexID(onlyJobVertex.getID(), 0);
        final ExecutionVertexID executionVertexId1 =
                new ExecutionVertexID(onlyJobVertex.getID(), 1);
        assertThat(testExecutionOperations.getDeployedVertices())
                .withFailMessage(
                        "The "
                                + "execution vertices should be deployed in a specific order reflecting the "
                                + "scheduling start and the global fail-over afterwards.")
                .contains(
                        executionVertexId0,
                        executionVertexId1,
                        executionVertexId0,
                        executionVertexId1);
    }