public FailureHandlingResult getGlobalFailureHandlingResult(
            final Throwable cause, long timestamp) {
        return handleFailureAndReport(
                null,
                cause,
                timestamp,
                IterableUtils.toStream(schedulingTopology.getVertices())
                        .map(SchedulingExecutionVertex::getId)
                        .collect(Collectors.toSet()),
                true);
    }
@Test
    void testGlobalFailureHandling() throws ExecutionException, InterruptedException {
        final Throwable error = new Exception("Expected test failure");
        final long timestamp = System.currentTimeMillis();
        final FailureHandlingResult result =
                executionFailureHandler.getGlobalFailureHandlingResult(error, timestamp);

        assertThat(result.getVerticesToRestart())
                .isEqualTo(
                        IterableUtils.toStream(schedulingTopology.getVertices())
                                .map(SchedulingExecutionVertex::getId)
                                .collect(Collectors.toSet()));
        assertThat(result.getError()).isSameAs(error);
        assertThat(result.getTimestamp()).isEqualTo(timestamp);
        assertThat(testingFailureEnricher.getSeenThrowables()).containsExactly(error);
        assertThat(result.getFailureLabels().get())
                .isEqualTo(testingFailureEnricher.getFailureLabels());
        checkMetrics(spanCollector, true, true);
    }