public static RootExceptionHistoryEntry fromFailureHandlingResultSnapshot(
            FailureHandlingResultSnapshot snapshot) {
        String failingTaskName = null;
        TaskManagerLocation taskManagerLocation = null;
        if (snapshot.getRootCauseExecution().isPresent()) {
            final Execution rootCauseExecution = snapshot.getRootCauseExecution().get();
            failingTaskName = rootCauseExecution.getVertexWithAttempt();
            taskManagerLocation = rootCauseExecution.getAssignedResourceLocation();
        }

        return createRootExceptionHistoryEntry(
                snapshot.getRootCause(),
                snapshot.getTimestamp(),
                snapshot.getFailureLabels(),
                failingTaskName,
                taskManagerLocation,
                snapshot.getConcurrentlyFailedExecution());
    }
@Test
    void testFromFailureHandlingResultSnapshot() throws ExecutionException, InterruptedException {
        final Throwable rootException = new RuntimeException("Expected root failure");
        final ExecutionVertex rootExecutionVertex = extractExecutionVertex(0);
        final long rootTimestamp = triggerFailure(rootExecutionVertex, rootException);
        final CompletableFuture<Map<String, String>> rootFailureLabels =
                CompletableFuture.completedFuture(Collections.singletonMap("key", "value"));

        final Throwable concurrentException1 = new IllegalStateException("Expected other failure1");
        final ExecutionVertex concurrentlyFailedExecutionVertex1 = extractExecutionVertex(1);
        Predicate<ExceptionHistoryEntry> exception1Predicate =
                triggerFailureAndCreateEntryMatcher(
                        concurrentException1, concurrentlyFailedExecutionVertex1);

        final FailureHandlingResultSnapshot snapshot =
                new FailureHandlingResultSnapshot(
                        rootExecutionVertex.getCurrentExecutionAttempt(),
                        rootException,
                        rootTimestamp,
                        rootFailureLabels,
                        Collections.singleton(
                                concurrentlyFailedExecutionVertex1.getCurrentExecutionAttempt()),
                        true);
        final RootExceptionHistoryEntry actualEntry =
                RootExceptionHistoryEntry.fromFailureHandlingResultSnapshot(snapshot);

        assertThat(actualEntry)
                .matches(
                        ExceptionHistoryEntryMatcher.matchesFailure(
                                rootException,
                                rootTimestamp,
                                rootFailureLabels.get(),
                                rootExecutionVertex.getTaskNameWithSubtaskIndex(),
                                rootExecutionVertex.getCurrentAssignedResourceLocation()));

        assertThat(actualEntry.getConcurrentExceptions()).hasSize(1).allMatch(exception1Predicate);

        // Test for addConcurrentExceptions
        final Throwable concurrentException2 = new IllegalStateException("Expected other failure2");
        final ExecutionVertex concurrentlyFailedExecutionVertex2 = extractExecutionVertex(2);
        Predicate<ExceptionHistoryEntry> exception2Predicate =
                triggerFailureAndCreateEntryMatcher(
                        concurrentException2, concurrentlyFailedExecutionVertex2);

        actualEntry.addConcurrentExceptions(
                concurrentlyFailedExecutionVertex2.getCurrentExecutions());
        assertThat(actualEntry.getConcurrentExceptions())
                .hasSize(2)
                .allMatch(
                        exceptionHistoryEntry ->
                                exception1Predicate.test(exceptionHistoryEntry)
                                        || exception2Predicate.test(exceptionHistoryEntry));
    }