public static ExceptionHistoryEntry create(
            AccessExecution failedExecution,
            String taskName,
            CompletableFuture<Map<String, String>> failureLabels) {
        Preconditions.checkNotNull(failedExecution, "No Execution is specified.");
        Preconditions.checkNotNull(taskName, "No task name is specified.");
        Preconditions.checkArgument(
                failedExecution.getFailureInfo().isPresent(),
                "The selected Execution " + failedExecution.getAttemptId() + " didn't fail.");

        final ErrorInfo failure = failedExecution.getFailureInfo().get();
        return new ExceptionHistoryEntry(
                failure.getException(),
                failure.getTimestamp(),
                failureLabels,
                taskName,
                failedExecution.getAssignedResourceLocation());
    }
@Test
    void testCreate() {
        final Throwable failure = new RuntimeException("Expected exception");
        final long timestamp = System.currentTimeMillis();
        final TaskManagerLocation taskManagerLocation = new LocalTaskManagerLocation();
        final AccessExecution execution =
                TestingAccessExecution.newBuilder()
                        .withErrorInfo(new ErrorInfo(failure, timestamp))
                        .withTaskManagerLocation(taskManagerLocation)
                        .build();
        final String taskName = "task name";
        final Map<String, String> failureLabels = Collections.singletonMap("key", "value");

        final ExceptionHistoryEntry entry =
                ExceptionHistoryEntry.create(
                        execution, taskName, CompletableFuture.completedFuture(failureLabels));

        assertThat(entry.getException().deserializeError(ClassLoader.getSystemClassLoader()))
                .isEqualTo(failure);
        assertThat(entry.getTimestamp()).isEqualTo(timestamp);
        assertThat(entry.getFailingTaskName()).isEqualTo(taskName);
        assertThat(entry.getTaskManagerLocation())
                .matches(isArchivedTaskManagerLocation(taskManagerLocation));
        assertThat(entry.isGlobal()).isFalse();
        assertThat(entry.getFailureLabels()).isEqualTo(failureLabels);
    }