@Test public void testJobResultRetrieval() throws Exception { final MiniDispatcher miniDispatcher = createMiniDispatcher(ClusterEntrypoint.ExecutionMode.NORMAL); miniDispatcher.start(); try { dispatcherLeaderElectionService.isLeader(UUID.randomUUID()).get(); jobGraphFuture.get(); resultFuture.complete(archivedExecutionGraph); 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, timeout); } }
@Override public CompletableFuture<JobResult> requestJobResult(JobID jobId, Time timeout) { final CompletableFuture<JobResult> jobResultFuture = super.requestJobResult(jobId, timeout); if (executionMode == ClusterEntrypoint.ExecutionMode.NORMAL) { jobResultFuture.thenAccept((JobResult result) -> { ApplicationStatus status = result.getSerializedThrowable().isPresent() ? ApplicationStatus.FAILED : ApplicationStatus.SUCCEEDED; jobTerminationFuture.complete(status); }); } return jobResultFuture; }
MiniDispatcher extends Dispatcher { @Override public CompletableFuture<JobResult> requestJobResult(JobID jobId, Time timeout) { final CompletableFuture<JobResult> jobResultFuture = super.requestJobResult(jobId, timeout); if (executionMode == ClusterEntrypoint.ExecutionMode.NORMAL) { jobResultFuture.thenAccept((JobResult result) -> { ApplicationStatus status = result.getSerializedThrowable().isPresent() ? ApplicationStatus.FAILED : ApplicationStatus.SUCCEEDED; jobTerminationFuture.complete(status); }); } return jobResultFuture; } }
MiniDispatcher extends Dispatcher { @Override public CompletableFuture<JobResult> requestJobResult(JobID jobId, Time timeout) { final CompletableFuture<JobResult> jobResultFuture = super.requestJobResult(jobId, timeout); if (executionMode == ClusterEntrypoint.ExecutionMode.NORMAL) { jobResultFuture.thenAccept((JobResult result) -> { ApplicationStatus status = result.getSerializedThrowable().isPresent() ? ApplicationStatus.FAILED : ApplicationStatus.SUCCEEDED; jobTerminationFuture.complete(status); }); } return jobResultFuture; } MiniDispatcher(
			RpcService rpcService,
			String endpointId,
			Configuration configuration,
			HighAvailabilityServices highAvailabilityServices,
			GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
			BlobServer blobServer,
			HeartbeatServices heartbeatServices,
			JobManagerMetricGroup jobManagerMetricGroup,
			@Nullable String metricQueryServiceAddress,
			ArchivedExecutionGraphStore archivedExecutionGraphStore,
			JobManagerRunnerFactory jobManagerRunnerFactory,
			FatalErrorHandler fatalErrorHandler,
			HistoryServerArchivist historyServerArchivist,
			JobGraph jobGraph,
			JobClusterEntrypoint.ExecutionMode executionMode); }
MiniDispatcher extends Dispatcher { @Override public CompletableFuture<JobResult> requestJobResult(JobID jobId, Time timeout) { final CompletableFuture<JobResult> jobResultFuture = super.requestJobResult(jobId, timeout); if (executionMode == ClusterEntrypoint.ExecutionMode.NORMAL) { jobResultFuture.thenAccept((JobResult result) -> { ApplicationStatus status = result.getSerializedThrowable().isPresent() ? ApplicationStatus.FAILED : ApplicationStatus.SUCCEEDED; jobTerminationFuture.complete(status); }); } return jobResultFuture; } MiniDispatcher(
			RpcService rpcService,
			String endpointId,
			Configuration configuration,
			HighAvailabilityServices highAvailabilityServices,
			GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
			BlobServer blobServer,
			HeartbeatServices heartbeatServices,
			JobManagerMetricGroup jobManagerMetricGroup,
			@Nullable String metricQueryServiceAddress,
			ArchivedExecutionGraphStore archivedExecutionGraphStore,
			JobManagerRunnerFactory jobManagerRunnerFactory,
			FatalErrorHandler fatalErrorHandler,
			HistoryServerArchivist historyServerArchivist,
			JobGraph jobGraph,
			JobClusterEntrypoint.ExecutionMode executionMode); CompletableFuture<ApplicationStatus> getJobTerminationFuture(); @Override CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout); @Override CompletableFuture<JobResult> requestJobResult(JobID jobId, Time timeout); }
MiniDispatcher extends Dispatcher { @Override public CompletableFuture<JobResult> requestJobResult(JobID jobId, Time timeout) { final CompletableFuture<JobResult> jobResultFuture = super.requestJobResult(jobId, timeout); if (executionMode == ClusterEntrypoint.ExecutionMode.NORMAL) { jobResultFuture.thenAccept((JobResult result) -> { ApplicationStatus status = result.getSerializedThrowable().isPresent() ? ApplicationStatus.FAILED : ApplicationStatus.SUCCEEDED; jobTerminationFuture.complete(status); }); } return jobResultFuture; } MiniDispatcher(
			RpcService rpcService,
			String endpointId,
			Configuration configuration,
			HighAvailabilityServices highAvailabilityServices,
			GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
			BlobServer blobServer,
			HeartbeatServices heartbeatServices,
			JobManagerMetricGroup jobManagerMetricGroup,
			@Nullable String metricQueryServiceAddress,
			ArchivedExecutionGraphStore archivedExecutionGraphStore,
			JobManagerRunnerFactory jobManagerRunnerFactory,
			FatalErrorHandler fatalErrorHandler,
			HistoryServerArchivist historyServerArchivist,
			JobGraph jobGraph,
			JobClusterEntrypoint.ExecutionMode executionMode); CompletableFuture<ApplicationStatus> getJobTerminationFuture(); @Override CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout); @Override CompletableFuture<JobResult> requestJobResult(JobID jobId, Time timeout);  }