@Test public void testOnAddedJobGraphRecoveryFailure() throws Exception { dispatcher = createAndStartDispatcher(heartbeatServices, haServices, new ExpectedJobIdJobManagerRunnerFactory(TEST_JOB_ID, createdJobManagerRunnerLatch)); final FlinkException expectedFailure = new FlinkException("Expected failure"); submittedJobGraphStore.setRecoveryFailure(expectedFailure); dispatcherLeaderElectionService.isLeader(UUID.randomUUID()).get(); submittedJobGraphStore.putJobGraph(new SubmittedJobGraph(jobGraph)); dispatcher.onAddedJobGraph(TEST_JOB_ID); final CompletableFuture<Throwable> errorFuture = fatalErrorHandler.getErrorFuture(); final Throwable throwable = errorFuture.get(); assertThat(ExceptionUtils.findThrowable(throwable, expectedFailure::equals).isPresent(), is(true)); fatalErrorHandler.clearError(); }
@Override public void onAddedJobGraph(final JobID jobId) { runAsync( () -> { if (!jobManagerRunnerFutures.containsKey(jobId)) { final CompletableFuture<Optional<JobGraph>> recoveredJob = recoveryOperation.thenApplyAsync( FunctionUtils.uncheckedFunction(ignored -> Optional.ofNullable(recoverJob(jobId))), getRpcService().getExecutor()); final DispatcherId dispatcherId = getFencingToken(); final CompletableFuture<Void> submissionFuture = recoveredJob.thenComposeAsync( (Optional<JobGraph> jobGraphOptional) -> jobGraphOptional.map( FunctionUtils.uncheckedFunction(jobGraph -> tryRunRecoveredJobGraph(jobGraph, dispatcherId).thenAcceptAsync( FunctionUtils.uncheckedConsumer((Boolean isRecoveredJobRunning) -> { if (!isRecoveredJobRunning) { submittedJobGraphStore.releaseJobGraph(jobId); } }), getRpcService().getExecutor()))) .orElse(CompletableFuture.completedFuture(null)), getUnfencedMainThreadExecutor()); submissionFuture.whenComplete( (Void ignored, Throwable throwable) -> { if (throwable != null) { onFatalError( new DispatcherException( String.format("Could not start the added job %s", jobId), ExceptionUtils.stripCompletionException(throwable))); } }); recoveryOperation = submissionFuture; } }); }
Dispatcher extends FencedRpcEndpoint<DispatcherId> implements
	DispatcherGateway, LeaderContender, SubmittedJobGraphStore.SubmittedJobGraphListener { @Override public void onAddedJobGraph(final JobID jobId) { runAsync( () -> { if (!jobManagerRunnerFutures.containsKey(jobId)) { final CompletableFuture<Optional<JobGraph>> recoveredJob = recoveryOperation.thenApplyAsync( FunctionUtils.uncheckedFunction(ignored -> Optional.ofNullable(recoverJob(jobId))), getRpcService().getExecutor()); final DispatcherId dispatcherId = getFencingToken(); final CompletableFuture<Void> submissionFuture = recoveredJob.thenComposeAsync( (Optional<JobGraph> jobGraphOptional) -> jobGraphOptional.map( FunctionUtils.uncheckedFunction(jobGraph -> tryRunRecoveredJobGraph(jobGraph, dispatcherId).thenAcceptAsync( FunctionUtils.uncheckedConsumer((Boolean isRecoveredJobRunning) -> { if (!isRecoveredJobRunning) { submittedJobGraphStore.releaseJobGraph(jobId); } }), getRpcService().getExecutor()))) .orElse(CompletableFuture.completedFuture(null)), getUnfencedMainThreadExecutor()); submissionFuture.whenComplete( (Void ignored, Throwable throwable) -> { if (throwable != null) { onFatalError( new DispatcherException( String.format("Could not start the added job %s", jobId), ExceptionUtils.stripCompletionException(throwable))); } }); recoveryOperation = submissionFuture; } }); } }
Dispatcher extends FencedRpcEndpoint<DispatcherId> implements
	DispatcherGateway, LeaderContender, SubmittedJobGraphStore.SubmittedJobGraphListener { @Override public void onAddedJobGraph(final JobID jobId) { runAsync( () -> { if (!jobManagerRunnerFutures.containsKey(jobId)) { final CompletableFuture<Optional<JobGraph>> recoveredJob = recoveryOperation.thenApplyAsync( FunctionUtils.uncheckedFunction(ignored -> Optional.ofNullable(recoverJob(jobId))), getRpcService().getExecutor()); final DispatcherId dispatcherId = getFencingToken(); final CompletableFuture<Void> submissionFuture = recoveredJob.thenComposeAsync( (Optional<JobGraph> jobGraphOptional) -> jobGraphOptional.map( FunctionUtils.uncheckedFunction(jobGraph -> tryRunRecoveredJobGraph(jobGraph, dispatcherId).thenAcceptAsync( FunctionUtils.uncheckedConsumer((Boolean isRecoveredJobRunning) -> { if (!isRecoveredJobRunning) { submittedJobGraphStore.releaseJobGraph(jobId); } }), getRpcService().getExecutor()))) .orElse(CompletableFuture.completedFuture(null)), getUnfencedMainThreadExecutor()); submissionFuture.whenComplete( (Void ignored, Throwable throwable) -> { if (throwable != null) { onFatalError( new DispatcherException( String.format("Could not start the added job %s", jobId), ExceptionUtils.stripCompletionException(throwable))); } }); recoveryOperation = submissionFuture; } }); } Dispatcher(
			RpcService rpcService,
			String endpointId,
			Configuration configuration,
			HighAvailabilityServices highAvailabilityServices,
			SubmittedJobGraphStore submittedJobGraphStore,
			GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
			BlobServer blobServer,
			HeartbeatServices heartbeatServices,
			JobManagerMetricGroup jobManagerMetricGroup,
			@Nullable String metricServiceQueryAddress,
			ArchivedExecutionGraphStore archivedExecutionGraphStore,
			JobManagerRunnerFactory jobManagerRunnerFactory,
			FatalErrorHandler fatalErrorHandler,
			HistoryServerArchivist historyServerArchivist); }
Dispatcher extends FencedRpcEndpoint<DispatcherId> implements
	DispatcherGateway, LeaderContender, SubmittedJobGraphStore.SubmittedJobGraphListener { @Override public void onAddedJobGraph(final JobID jobId) { runAsync( () -> { if (!jobManagerRunnerFutures.containsKey(jobId)) { final CompletableFuture<Optional<JobGraph>> recoveredJob = recoveryOperation.thenApplyAsync( FunctionUtils.uncheckedFunction(ignored -> Optional.ofNullable(recoverJob(jobId))), getRpcService().getExecutor()); final DispatcherId dispatcherId = getFencingToken(); final CompletableFuture<Void> submissionFuture = recoveredJob.thenComposeAsync( (Optional<JobGraph> jobGraphOptional) -> jobGraphOptional.map( FunctionUtils.uncheckedFunction(jobGraph -> tryRunRecoveredJobGraph(jobGraph, dispatcherId).thenAcceptAsync( FunctionUtils.uncheckedConsumer((Boolean isRecoveredJobRunning) -> { if (!isRecoveredJobRunning) { submittedJobGraphStore.releaseJobGraph(jobId); } }), getRpcService().getExecutor()))) .orElse(CompletableFuture.completedFuture(null)), getUnfencedMainThreadExecutor()); submissionFuture.whenComplete( (Void ignored, Throwable throwable) -> { if (throwable != null) { onFatalError( new DispatcherException( String.format("Could not start the added job %s", jobId), ExceptionUtils.stripCompletionException(throwable))); } }); recoveryOperation = submissionFuture; } }); } Dispatcher(
			RpcService rpcService,
			String endpointId,
			Configuration configuration,
			HighAvailabilityServices highAvailabilityServices,
			SubmittedJobGraphStore submittedJobGraphStore,
			GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
			BlobServer blobServer,
			HeartbeatServices heartbeatServices,
			JobManagerMetricGroup jobManagerMetricGroup,
			@Nullable String metricServiceQueryAddress,
			ArchivedExecutionGraphStore archivedExecutionGraphStore,
			JobManagerRunnerFactory jobManagerRunnerFactory,
			FatalErrorHandler fatalErrorHandler,
			HistoryServerArchivist historyServerArchivist); @Override void onStart(); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout); @Override CompletableFuture<Collection<JobID>> listJobs(Time timeout); @Override CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath, Time timeout); @Override CompletableFuture<Acknowledge> cancelJob(JobID jobId, Time timeout); @Override CompletableFuture<ClusterOverview> requestClusterOverview(Time timeout); @Override CompletableFuture<MultipleJobsDetails> requestMultipleJobDetails(Time timeout); @Override CompletableFuture<JobStatus> requestJobStatus(JobID jobId, Time timeout); @Override CompletableFuture<OperatorBackPressureStatsResponse> requestOperatorBackPressureStats(
			final JobID jobId,
			final JobVertexID jobVertexId); @Override CompletableFuture<ArchivedExecutionGraph> requestJob(JobID jobId, Time timeout); @Override CompletableFuture<JobResult> requestJobResult(JobID jobId, Time timeout); @Override CompletableFuture<Collection<String>> requestMetricQueryServiceAddresses(Time timeout); @Override CompletableFuture<Collection<Tuple2<ResourceID, String>>> requestTaskManagerMetricQueryServiceAddresses(Time timeout); @Override CompletableFuture<Integer> getBlobServerPort(Time timeout); @Override CompletableFuture<String> triggerSavepoint(
			final JobID jobId,
			final String targetDirectory,
			final boolean cancelJob,
			final Time timeout); @Override CompletableFuture<String> stopWithSavepoint(
			final JobID jobId,
			final String targetDirectory,
			final boolean advanceToEndOfEventTime,
			final Time timeout); @Override CompletableFuture<Acknowledge> shutDownCluster(); @Override void grantLeadership(final UUID newLeaderSessionID); @Override void revokeLeadership(); @Override void handleError(final Exception exception); @Override void onAddedJobGraph(final JobID jobId); @Override void onRemovedJobGraph(final JobID jobId); }
Dispatcher extends FencedRpcEndpoint<DispatcherId> implements
	DispatcherGateway, LeaderContender, SubmittedJobGraphStore.SubmittedJobGraphListener { @Override public void onAddedJobGraph(final JobID jobId) { runAsync( () -> { if (!jobManagerRunnerFutures.containsKey(jobId)) { final CompletableFuture<Optional<JobGraph>> recoveredJob = recoveryOperation.thenApplyAsync( FunctionUtils.uncheckedFunction(ignored -> Optional.ofNullable(recoverJob(jobId))), getRpcService().getExecutor()); final DispatcherId dispatcherId = getFencingToken(); final CompletableFuture<Void> submissionFuture = recoveredJob.thenComposeAsync( (Optional<JobGraph> jobGraphOptional) -> jobGraphOptional.map( FunctionUtils.uncheckedFunction(jobGraph -> tryRunRecoveredJobGraph(jobGraph, dispatcherId).thenAcceptAsync( FunctionUtils.uncheckedConsumer((Boolean isRecoveredJobRunning) -> { if (!isRecoveredJobRunning) { submittedJobGraphStore.releaseJobGraph(jobId); } }), getRpcService().getExecutor()))) .orElse(CompletableFuture.completedFuture(null)), getUnfencedMainThreadExecutor()); submissionFuture.whenComplete( (Void ignored, Throwable throwable) -> { if (throwable != null) { onFatalError( new DispatcherException( String.format("Could not start the added job %s", jobId), ExceptionUtils.stripCompletionException(throwable))); } }); recoveryOperation = submissionFuture; } }); } Dispatcher(
			RpcService rpcService,
			String endpointId,
			Configuration configuration,
			HighAvailabilityServices highAvailabilityServices,
			SubmittedJobGraphStore submittedJobGraphStore,
			GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
			BlobServer blobServer,
			HeartbeatServices heartbeatServices,
			JobManagerMetricGroup jobManagerMetricGroup,
			@Nullable String metricServiceQueryAddress,
			ArchivedExecutionGraphStore archivedExecutionGraphStore,
			JobManagerRunnerFactory jobManagerRunnerFactory,
			FatalErrorHandler fatalErrorHandler,
			HistoryServerArchivist historyServerArchivist); @Override void onStart(); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout); @Override CompletableFuture<Collection<JobID>> listJobs(Time timeout); @Override CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath, Time timeout); @Override CompletableFuture<Acknowledge> cancelJob(JobID jobId, Time timeout); @Override CompletableFuture<ClusterOverview> requestClusterOverview(Time timeout); @Override CompletableFuture<MultipleJobsDetails> requestMultipleJobDetails(Time timeout); @Override CompletableFuture<JobStatus> requestJobStatus(JobID jobId, Time timeout); @Override CompletableFuture<OperatorBackPressureStatsResponse> requestOperatorBackPressureStats(
			final JobID jobId,
			final JobVertexID jobVertexId); @Override CompletableFuture<ArchivedExecutionGraph> requestJob(JobID jobId, Time timeout); @Override CompletableFuture<JobResult> requestJobResult(JobID jobId, Time timeout); @Override CompletableFuture<Collection<String>> requestMetricQueryServiceAddresses(Time timeout); @Override CompletableFuture<Collection<Tuple2<ResourceID, String>>> requestTaskManagerMetricQueryServiceAddresses(Time timeout); @Override CompletableFuture<Integer> getBlobServerPort(Time timeout); @Override CompletableFuture<String> triggerSavepoint(
			final JobID jobId,
			final String targetDirectory,
			final boolean cancelJob,
			final Time timeout); @Override CompletableFuture<String> stopWithSavepoint(
			final JobID jobId,
			final String targetDirectory,
			final boolean advanceToEndOfEventTime,
			final Time timeout); @Override CompletableFuture<Acknowledge> shutDownCluster(); @Override void grantLeadership(final UUID newLeaderSessionID); @Override void revokeLeadership(); @Override void handleError(final Exception exception); @Override void onAddedJobGraph(final JobID jobId); @Override void onRemovedJobGraph(final JobID jobId); static final String DISPATCHER_NAME; }