@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 handleGlobalFailure() {
        final JobGraph jobGraph = singleNonParallelJobVertexJobGraph();
        final JobVertex onlyJobVertex = getOnlyJobVertex(jobGraph);

        final DefaultScheduler scheduler = createSchedulerAndStartScheduling(jobGraph);

        scheduler.handleGlobalFailure(new Exception("forced failure"));

        final ArchivedExecutionVertex onlyExecutionVertex =
                Iterables.getOnlyElement(
                        scheduler
                                .requestJob()
                                .getArchivedExecutionGraph()
                                .getAllExecutionVertices());
        final ExecutionAttemptID attemptId =
                onlyExecutionVertex.getCurrentExecutionAttempt().getAttemptId();
        scheduler.updateTaskExecutionState(
                new TaskExecutionState(attemptId, ExecutionState.CANCELED));

        taskRestartExecutor.triggerScheduledTasks();

        final List<ExecutionVertexID> deployedExecutionVertices =
                testExecutionOperations.getDeployedVertices();
        final ExecutionVertexID executionVertexId = new ExecutionVertexID(onlyJobVertex.getID(), 0);
        assertThat(deployedExecutionVertices).contains(executionVertexId, executionVertexId);
    }