@Test public void testJobSubmissionWithPartialResourceConfigured() throws Exception { dispatcher = createAndStartDispatcher(heartbeatServices, haServices, new ExpectedJobIdJobManagerRunnerFactory(TEST_JOB_ID, createdJobManagerRunnerLatch)); CompletableFuture<UUID> leaderFuture = dispatcherLeaderElectionService.isLeader(UUID.randomUUID()); leaderFuture.get(); DispatcherGateway dispatcherGateway = dispatcher.getSelfGateway(DispatcherGateway.class); ResourceSpec resourceSpec = ResourceSpec.newBuilder().setCpuCores(2).build(); final JobVertex firstVertex = new JobVertex("firstVertex"); firstVertex.setInvokableClass(NoOpInvokable.class); firstVertex.setResources(resourceSpec, resourceSpec); final JobVertex secondVertex = new JobVertex("secondVertex"); secondVertex.setInvokableClass(NoOpInvokable.class); JobGraph jobGraphWithTwoVertices = new JobGraph(TEST_JOB_ID, "twoVerticesJob", firstVertex, secondVertex); jobGraphWithTwoVertices.setAllowQueuedScheduling(true); CompletableFuture<Acknowledge> acknowledgeFuture = dispatcherGateway.submitJob(jobGraphWithTwoVertices, TIMEOUT); try { acknowledgeFuture.get(); fail("job submission should have failed"); } catch (ExecutionException e) { assertTrue(ExceptionUtils.findThrowable(e, JobSubmissionException.class).isPresent()); } }
@Override public CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout) { log.info("Received JobGraph submission {} ({}).", jobGraph.getJobID(), jobGraph.getName()); try { if (isDuplicateJob(jobGraph.getJobID())) { return FutureUtils.completedExceptionally( new JobSubmissionException(jobGraph.getJobID(), "Job has already been submitted.")); } else if (isPartialResourceConfigured(jobGraph)) { return FutureUtils.completedExceptionally( new JobSubmissionException(jobGraph.getJobID(), "Currently jobs is not supported if parts of the vertices have " + "resources configured. The limitation will be removed in future versions.")); } else { return internalSubmitJob(jobGraph); } } catch (FlinkException e) { return FutureUtils.completedExceptionally(e); } }
Dispatcher extends FencedRpcEndpoint<DispatcherId> implements
	DispatcherGateway, LeaderContender, SubmittedJobGraphStore.SubmittedJobGraphListener { @Override public CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout) { log.info("Received JobGraph submission {} ({}).", jobGraph.getJobID(), jobGraph.getName()); try { if (isDuplicateJob(jobGraph.getJobID())) { return FutureUtils.completedExceptionally( new JobSubmissionException(jobGraph.getJobID(), "Job has already been submitted.")); } else if (isPartialResourceConfigured(jobGraph)) { return FutureUtils.completedExceptionally( new JobSubmissionException(jobGraph.getJobID(), "Currently jobs is not supported if parts of the vertices have " + "resources configured. The limitation will be removed in future versions.")); } else { return internalSubmitJob(jobGraph); } } catch (FlinkException e) { return FutureUtils.completedExceptionally(e); } } }
Dispatcher extends FencedRpcEndpoint<DispatcherId> implements
	DispatcherGateway, LeaderContender, SubmittedJobGraphStore.SubmittedJobGraphListener { @Override public CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout) { log.info("Received JobGraph submission {} ({}).", jobGraph.getJobID(), jobGraph.getName()); try { if (isDuplicateJob(jobGraph.getJobID())) { return FutureUtils.completedExceptionally( new JobSubmissionException(jobGraph.getJobID(), "Job has already been submitted.")); } else if (isPartialResourceConfigured(jobGraph)) { return FutureUtils.completedExceptionally( new JobSubmissionException(jobGraph.getJobID(), "Currently jobs is not supported if parts of the vertices have " + "resources configured. The limitation will be removed in future versions.")); } else { return internalSubmitJob(jobGraph); } } catch (FlinkException e) { return FutureUtils.completedExceptionally(e); } } 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 CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout) { log.info("Received JobGraph submission {} ({}).", jobGraph.getJobID(), jobGraph.getName()); try { if (isDuplicateJob(jobGraph.getJobID())) { return FutureUtils.completedExceptionally( new JobSubmissionException(jobGraph.getJobID(), "Job has already been submitted.")); } else if (isPartialResourceConfigured(jobGraph)) { return FutureUtils.completedExceptionally( new JobSubmissionException(jobGraph.getJobID(), "Currently jobs is not supported if parts of the vertices have " + "resources configured. The limitation will be removed in future versions.")); } else { return internalSubmitJob(jobGraph); } } catch (FlinkException e) { return FutureUtils.completedExceptionally(e); } } 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 CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout) { log.info("Received JobGraph submission {} ({}).", jobGraph.getJobID(), jobGraph.getName()); try { if (isDuplicateJob(jobGraph.getJobID())) { return FutureUtils.completedExceptionally( new JobSubmissionException(jobGraph.getJobID(), "Job has already been submitted.")); } else if (isPartialResourceConfigured(jobGraph)) { return FutureUtils.completedExceptionally( new JobSubmissionException(jobGraph.getJobID(), "Currently jobs is not supported if parts of the vertices have " + "resources configured. The limitation will be removed in future versions.")); } else { return internalSubmitJob(jobGraph); } } catch (FlinkException e) { return FutureUtils.completedExceptionally(e); } } 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; }