@Override
    public CompletableFuture<Acknowledge> submitFailedJob(
            JobID jobId, String jobName, Throwable exception) {
        final ArchivedExecutionGraph archivedExecutionGraph =
                ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                        jobId,
                        jobName,
                        JobStatus.FAILED,
                        null,
                        exception,
                        null,
                        System.currentTimeMillis());
        ExecutionGraphInfo executionGraphInfo = new ExecutionGraphInfo(archivedExecutionGraph);
        writeToExecutionGraphInfoStore(executionGraphInfo);
        return archiveExecutionGraphToHistoryServer(executionGraphInfo);
    }
@Test
    public void testRetrieveJobResultAfterSubmissionOfFailedJob() throws Exception {
        dispatcher =
                createAndStartDispatcher(
                        heartbeatServices,
                        haServices,
                        new ExpectedJobIdJobManagerRunnerFactory(jobId));
        final DispatcherGateway dispatcherGateway =
                dispatcher.getSelfGateway(DispatcherGateway.class);
        final JobID failedJobId = new JobID();
        final String failedJobName = "test";
        final CompletableFuture<Acknowledge> submitFuture =
                dispatcherGateway.submitFailedJob(
                        failedJobId, failedJobName, new RuntimeException("Test exception."));
        submitFuture.get();
        final ArchivedExecutionGraph archivedExecutionGraph =
                dispatcherGateway.requestJob(failedJobId, TIMEOUT).get();
        Assertions.assertThat(archivedExecutionGraph.getJobID()).isEqualTo(failedJobId);
        Assertions.assertThat(archivedExecutionGraph.getJobName()).isEqualTo(failedJobName);
        Assertions.assertThat(archivedExecutionGraph.getState()).isEqualTo(JobStatus.FAILED);
        Assertions.assertThat(archivedExecutionGraph.getFailureInfo())
                .isNotNull()
                .extracting(ErrorInfo::getException)
                .extracting(e -> e.deserializeError(Thread.currentThread().getContextClassLoader()))
                .satisfies(
                        exception ->
                                Assertions.assertThat(exception)
                                        .isInstanceOf(RuntimeException.class)
                                        .hasMessage("Test exception."));
    }