@Override
    public CompletableFuture<JobManagerRunnerResult> getResultFuture() {
        return resultFuture;
    }
@Test
    void testJobMasterTerminationIsHandled() {
        final CompletableFuture<JobMasterService> jobMasterServiceFuture =
                new CompletableFuture<>();
        DefaultJobMasterServiceProcess serviceProcess = createTestInstance(jobMasterServiceFuture);
        CompletableFuture<Void> jobMasterTerminationFuture = new CompletableFuture<>();
        TestingJobMasterService testingJobMasterService =
                new TestingJobMasterService("localhost", jobMasterTerminationFuture, null);
        jobMasterServiceFuture.complete(testingJobMasterService);

        RuntimeException testException = new RuntimeException("Fake exception from JobMaster");
        jobMasterTerminationFuture.completeExceptionally(testException);

        assertThatFuture(serviceProcess.getResultFuture())
                .eventuallyFailsWith(ExecutionException.class)
                .extracting(FlinkAssertions::chainOfCauses, FlinkAssertions.STREAM_THROWABLE)
                .anySatisfy(t -> assertThat(t).isEqualTo(testException));
    }