@Override
    public CompletableFuture<JobResult> requestJobResult(JobID jobId, Time timeout) {
        final CompletableFuture<JobResult> jobResultFuture = super.requestJobResult(jobId, timeout);

        if (executionMode == ClusterEntrypoint.ExecutionMode.NORMAL) {
            // terminate the MiniDispatcher once we served the first JobResult successfully
            jobResultFuture.thenAccept(
                    (JobResult result) -> {
                        ApplicationStatus status =
                                result.getSerializedThrowable().isPresent()
                                        ? ApplicationStatus.FAILED
                                        : ApplicationStatus.SUCCEEDED;

                        if (!ApplicationStatus.UNKNOWN.equals(result.getApplicationStatus())) {
                            log.info(
                                    "Shutting down cluster because someone retrieved the job result"
                                            + " and the status is globally terminal.");
                            shutDownFuture.complete(status);
                        }
                    });
        } else {
            log.info("Not shutting down cluster after someone retrieved the job result.");
        }

        return jobResultFuture;
    }
@Test
    public void testJobResultRetrieval() throws Exception {
        final MiniDispatcher miniDispatcher =
                createMiniDispatcher(ClusterEntrypoint.ExecutionMode.NORMAL);

        miniDispatcher.start();

        try {
            // wait until we have submitted the job
            final TestingJobManagerRunner testingJobManagerRunner =
                    testingJobManagerRunnerFactory.takeCreatedJobManagerRunner();

            testingJobManagerRunner.completeResultFuture(executionGraphInfo);

            assertFalse(miniDispatcher.getTerminationFuture().isDone());

            final DispatcherGateway dispatcherGateway =
                    miniDispatcher.getSelfGateway(DispatcherGateway.class);

            final CompletableFuture<JobResult> jobResultFuture =
                    dispatcherGateway.requestJobResult(jobGraph.getJobID(), timeout);

            final JobResult jobResult = jobResultFuture.get();

            assertThat(jobResult.getJobId(), is(jobGraph.getJobID()));
        } finally {
            RpcUtils.terminateRpcEndpoint(miniDispatcher);
        }
    }