public static FailureHandlingResultSnapshot create(
            FailureHandlingResult failureHandlingResult,
            Function<ExecutionVertexID, Collection<Execution>> currentExecutionsLookup) {
        final Execution rootCauseExecution =
                failureHandlingResult.getFailedExecution().orElse(null);

        if (rootCauseExecution != null && !rootCauseExecution.getFailureInfo().isPresent()) {
            throw new IllegalArgumentException(
                    String.format(
                            "The failed execution %s didn't provide a failure info.",
                            rootCauseExecution.getAttemptId()));
        }

        final Set<Execution> concurrentlyFailedExecutions =
                failureHandlingResult.getVerticesToRestart().stream()
                        .flatMap(id -> currentExecutionsLookup.apply(id).stream())
                        .filter(execution -> execution != rootCauseExecution)
                        .filter(execution -> execution.getFailureInfo().isPresent())
                        .collect(Collectors.toSet());

        return new FailureHandlingResultSnapshot(
                rootCauseExecution,
                ErrorInfo.handleMissingThrowable(failureHandlingResult.getError()),
                failureHandlingResult.getTimestamp(),
                failureHandlingResult.getFailureLabels(),
                concurrentlyFailedExecutions,
                failureHandlingResult.isRootCause());
    }
@Test
    void testRootCauseVertexNotFailed() {
        final ExecutionVertex rootCauseExecutionVertex = extractExecutionVertex(0);
        final FailureHandlingResult failureHandlingResult =
                FailureHandlingResult.restartable(
                        rootCauseExecutionVertex.getCurrentExecutionAttempt(),
                        new RuntimeException("Expected exception: root cause"),
                        System.currentTimeMillis(),
                        FailureEnricherUtils.EMPTY_FAILURE_LABELS,
                        StreamSupport.stream(
                                        executionGraph.getAllExecutionVertices().spliterator(),
                                        false)
                                .map(ExecutionVertex::getID)
                                .collect(Collectors.toSet()),
                        0L,
                        false,
                        true);

        assertThatThrownBy(
                        () ->
                                FailureHandlingResultSnapshot.create(
                                        failureHandlingResult, this::getCurrentExecutions))
                .isInstanceOf(IllegalArgumentException.class);
    }