@Test public void testRequestPartitionState() throws Exception { final JobGraph producerConsumerJobGraph = producerConsumerJobGraph(); final JobMaster jobMaster = createJobMaster( configuration, producerConsumerJobGraph, haServices, new TestingJobManagerSharedServicesBuilder().build(), heartbeatServices); CompletableFuture<Acknowledge> startFuture = jobMaster.start(jobMasterId); try { startFuture.get(testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS); final CompletableFuture<TaskDeploymentDescriptor> tddFuture = new CompletableFuture<>(); final TestingTaskExecutorGateway testingTaskExecutorGateway = new TestingTaskExecutorGatewayBuilder() .setSubmitTaskConsumer((taskDeploymentDescriptor, jobMasterId) -> { tddFuture.complete(taskDeploymentDescriptor); return CompletableFuture.completedFuture(Acknowledge.get()); }) .createTestingTaskExecutorGateway(); final JobMasterGateway jobMasterGateway = jobMaster.getSelfGateway(JobMasterGateway.class); final Collection<SlotOffer> slotOffers = registerSlotsAtJobMaster(1, jobMasterGateway, testingTaskExecutorGateway); assertThat(slotOffers, hasSize(1)); final TaskDeploymentDescriptor tdd = tddFuture.get(); assertThat(tdd.getProducedPartitions(), hasSize(1)); final ResultPartitionDeploymentDescriptor partition = tdd.getProducedPartitions().iterator().next(); final ExecutionAttemptID executionAttemptId = tdd.getExecutionAttemptId(); final ExecutionAttemptID copiedExecutionAttemptId = new ExecutionAttemptID(executionAttemptId.getLowerPart(), executionAttemptId.getUpperPart()); jobMasterGateway.updateTaskExecutionState(new TaskExecutionState(producerConsumerJobGraph.getJobID(), executionAttemptId, ExecutionState.FINISHED)).get(); final ResultPartitionID partitionId = new ResultPartitionID(partition.getPartitionId(), copiedExecutionAttemptId); CompletableFuture<ExecutionState> partitionStateFuture = jobMasterGateway.requestPartitionState(partition.getResultId(), partitionId); assertThat(partitionStateFuture.get(), equalTo(ExecutionState.FINISHED)); partitionStateFuture = jobMasterGateway.requestPartitionState(partition.getResultId(), new ResultPartitionID()); try { partitionStateFuture.get(); fail("Expected failure."); } catch (ExecutionException e) { assertThat(ExceptionUtils.findThrowable(e, IllegalArgumentException.class).isPresent(), is(true)); } partitionStateFuture = jobMasterGateway.requestPartitionState(new IntermediateDataSetID(), partitionId); try { partitionStateFuture.get(); fail("Expected failure."); } catch (ExecutionException e) { assertThat(ExceptionUtils.findThrowable(e, IllegalArgumentException.class).isPresent(), is(true)); } partitionStateFuture = jobMasterGateway.requestPartitionState(partition.getResultId(), new ResultPartitionID(partition.getPartitionId(), new ExecutionAttemptID())); try { partitionStateFuture.get(); fail("Expected failure."); } catch (ExecutionException e) { assertThat(ExceptionUtils.findThrowable(e, PartitionProducerDisposedException.class).isPresent(), is(true)); } } finally { RpcUtils.terminateRpcEndpoint(jobMaster, testingTimeout); } }
@Override public CompletableFuture<ExecutionState> requestPartitionState( final IntermediateDataSetID intermediateResultId, final ResultPartitionID resultPartitionId) { final Execution execution = executionGraph.getRegisteredExecutions().get(resultPartitionId.getProducerId()); if (execution != null) { return CompletableFuture.completedFuture(execution.getState()); } else { final IntermediateResult intermediateResult = executionGraph.getAllIntermediateResults().get(intermediateResultId); if (intermediateResult != null) { Execution producerExecution = intermediateResult .getPartitionById(resultPartitionId.getPartitionId()) .getProducer() .getCurrentExecutionAttempt(); if (producerExecution.getAttemptId().equals(resultPartitionId.getProducerId())) { return CompletableFuture.completedFuture(producerExecution.getState()); } else { return FutureUtils.completedExceptionally(new PartitionProducerDisposedException(resultPartitionId)); } } else { return FutureUtils.completedExceptionally(new IllegalArgumentException("Intermediate data set with ID " + intermediateResultId + " not found.")); } } }
JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway, JobMasterService { @Override public CompletableFuture<ExecutionState> requestPartitionState( final IntermediateDataSetID intermediateResultId, final ResultPartitionID resultPartitionId) { final Execution execution = executionGraph.getRegisteredExecutions().get(resultPartitionId.getProducerId()); if (execution != null) { return CompletableFuture.completedFuture(execution.getState()); } else { final IntermediateResult intermediateResult = executionGraph.getAllIntermediateResults().get(intermediateResultId); if (intermediateResult != null) { Execution producerExecution = intermediateResult .getPartitionById(resultPartitionId.getPartitionId()) .getProducer() .getCurrentExecutionAttempt(); if (producerExecution.getAttemptId().equals(resultPartitionId.getProducerId())) { return CompletableFuture.completedFuture(producerExecution.getState()); } else { return FutureUtils.completedExceptionally(new PartitionProducerDisposedException(resultPartitionId)); } } else { return FutureUtils.completedExceptionally(new IllegalArgumentException("Intermediate data set with ID " + intermediateResultId + " not found.")); } } } }
JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway, JobMasterService { @Override public CompletableFuture<ExecutionState> requestPartitionState( final IntermediateDataSetID intermediateResultId, final ResultPartitionID resultPartitionId) { final Execution execution = executionGraph.getRegisteredExecutions().get(resultPartitionId.getProducerId()); if (execution != null) { return CompletableFuture.completedFuture(execution.getState()); } else { final IntermediateResult intermediateResult = executionGraph.getAllIntermediateResults().get(intermediateResultId); if (intermediateResult != null) { Execution producerExecution = intermediateResult .getPartitionById(resultPartitionId.getPartitionId()) .getProducer() .getCurrentExecutionAttempt(); if (producerExecution.getAttemptId().equals(resultPartitionId.getProducerId())) { return CompletableFuture.completedFuture(producerExecution.getState()); } else { return FutureUtils.completedExceptionally(new PartitionProducerDisposedException(resultPartitionId)); } } else { return FutureUtils.completedExceptionally(new IllegalArgumentException("Intermediate data set with ID " + intermediateResultId + " not found.")); } } } JobMaster(
			RpcService rpcService,
			JobMasterConfiguration jobMasterConfiguration,
			ResourceID resourceId,
			JobGraph jobGraph,
			HighAvailabilityServices highAvailabilityService,
			SlotPoolFactory slotPoolFactory,
			SchedulerFactory schedulerFactory,
			JobManagerSharedServices jobManagerSharedServices,
			HeartbeatServices heartbeatServices,
			JobManagerJobMetricGroupFactory jobMetricGroupFactory,
			OnCompletionActions jobCompletionActions,
			FatalErrorHandler fatalErrorHandler,
			ClassLoader userCodeLoader); }
JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway, JobMasterService { @Override public CompletableFuture<ExecutionState> requestPartitionState( final IntermediateDataSetID intermediateResultId, final ResultPartitionID resultPartitionId) { final Execution execution = executionGraph.getRegisteredExecutions().get(resultPartitionId.getProducerId()); if (execution != null) { return CompletableFuture.completedFuture(execution.getState()); } else { final IntermediateResult intermediateResult = executionGraph.getAllIntermediateResults().get(intermediateResultId); if (intermediateResult != null) { Execution producerExecution = intermediateResult .getPartitionById(resultPartitionId.getPartitionId()) .getProducer() .getCurrentExecutionAttempt(); if (producerExecution.getAttemptId().equals(resultPartitionId.getProducerId())) { return CompletableFuture.completedFuture(producerExecution.getState()); } else { return FutureUtils.completedExceptionally(new PartitionProducerDisposedException(resultPartitionId)); } } else { return FutureUtils.completedExceptionally(new IllegalArgumentException("Intermediate data set with ID " + intermediateResultId + " not found.")); } } } JobMaster(
			RpcService rpcService,
			JobMasterConfiguration jobMasterConfiguration,
			ResourceID resourceId,
			JobGraph jobGraph,
			HighAvailabilityServices highAvailabilityService,
			SlotPoolFactory slotPoolFactory,
			SchedulerFactory schedulerFactory,
			JobManagerSharedServices jobManagerSharedServices,
			HeartbeatServices heartbeatServices,
			JobManagerJobMetricGroupFactory jobMetricGroupFactory,
			OnCompletionActions jobCompletionActions,
			FatalErrorHandler fatalErrorHandler,
			ClassLoader userCodeLoader); CompletableFuture<Acknowledge> start(final JobMasterId newJobMasterId); CompletableFuture<Acknowledge> suspend(final Exception cause); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<Acknowledge> cancel(Time timeout); @Override CompletableFuture<Acknowledge> stop(Time timeout); @Override CompletableFuture<Acknowledge> rescaleJob(
			int newParallelism,
			RescalingBehaviour rescalingBehaviour,
			Time timeout); @Override CompletableFuture<Acknowledge> rescaleOperators(
			Collection<JobVertexID> operators,
			int newParallelism,
			RescalingBehaviour rescalingBehaviour,
			Time timeout); @Override CompletableFuture<Acknowledge> updateTaskExecutionState(
			final TaskExecutionState taskExecutionState); @Override CompletableFuture<SerializedInputSplit> requestNextInputSplit(
			final JobVertexID vertexID,
			final ExecutionAttemptID executionAttempt); @Override CompletableFuture<ExecutionState> requestPartitionState(
			final IntermediateDataSetID intermediateResultId,
			final ResultPartitionID resultPartitionId); @Override CompletableFuture<Acknowledge> scheduleOrUpdateConsumers(
			final ResultPartitionID partitionID,
			final Time timeout); @Override CompletableFuture<Acknowledge> disconnectTaskManager(final ResourceID resourceID, final Exception cause); @Override void acknowledgeCheckpoint(
			final JobID jobID,
			final ExecutionAttemptID executionAttemptID,
			final long checkpointId,
			final CheckpointMetrics checkpointMetrics,
			final TaskStateSnapshot checkpointState); @Override void declineCheckpoint(DeclineCheckpoint decline); @Override CompletableFuture<KvStateLocation> requestKvStateLocation(final JobID jobId, final String registrationName); @Override CompletableFuture<Acknowledge> notifyKvStateRegistered(
			final JobID jobId,
			final JobVertexID jobVertexId,
			final KeyGroupRange keyGroupRange,
			final String registrationName,
			final KvStateID kvStateId,
			final InetSocketAddress kvStateServerAddress); @Override CompletableFuture<Acknowledge> notifyKvStateUnregistered(
			JobID jobId,
			JobVertexID jobVertexId,
			KeyGroupRange keyGroupRange,
			String registrationName); @Override CompletableFuture<Collection<SlotOffer>> offerSlots(
			final ResourceID taskManagerId,
			final Collection<SlotOffer> slots,
			final Time timeout); @Override void failSlot(
			final ResourceID taskManagerId,
			final AllocationID allocationId,
			final Exception cause); @Override CompletableFuture<RegistrationResponse> registerTaskManager(
			final String taskManagerRpcAddress,
			final TaskManagerLocation taskManagerLocation,
			final Time timeout); @Override void disconnectResourceManager(
			final ResourceManagerId resourceManagerId,
			final Exception cause); @Override void heartbeatFromTaskManager(final ResourceID resourceID, AccumulatorReport accumulatorReport); @Override void heartbeatFromResourceManager(final ResourceID resourceID); @Override CompletableFuture<JobDetails> requestJobDetails(Time timeout); @Override CompletableFuture<JobStatus> requestJobStatus(Time timeout); @Override CompletableFuture<ArchivedExecutionGraph> requestJob(Time timeout); @Override CompletableFuture<String> triggerSavepoint(
			@Nullable final String targetDirectory,
			final boolean cancelJob,
			final Time timeout); @Override CompletableFuture<OperatorBackPressureStatsResponse> requestOperatorBackPressureStats(final JobVertexID jobVertexId); @Override void notifyAllocationFailure(AllocationID allocationID, Exception cause); @Override CompletableFuture<Object> updateGlobalAggregate(String aggregateName, Object aggregand, byte[] serializedAggregateFunction); @Override JobMasterGateway getGateway(); }
JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway, JobMasterService { @Override public CompletableFuture<ExecutionState> requestPartitionState( final IntermediateDataSetID intermediateResultId, final ResultPartitionID resultPartitionId) { final Execution execution = executionGraph.getRegisteredExecutions().get(resultPartitionId.getProducerId()); if (execution != null) { return CompletableFuture.completedFuture(execution.getState()); } else { final IntermediateResult intermediateResult = executionGraph.getAllIntermediateResults().get(intermediateResultId); if (intermediateResult != null) { Execution producerExecution = intermediateResult .getPartitionById(resultPartitionId.getPartitionId()) .getProducer() .getCurrentExecutionAttempt(); if (producerExecution.getAttemptId().equals(resultPartitionId.getProducerId())) { return CompletableFuture.completedFuture(producerExecution.getState()); } else { return FutureUtils.completedExceptionally(new PartitionProducerDisposedException(resultPartitionId)); } } else { return FutureUtils.completedExceptionally(new IllegalArgumentException("Intermediate data set with ID " + intermediateResultId + " not found.")); } } } JobMaster(
			RpcService rpcService,
			JobMasterConfiguration jobMasterConfiguration,
			ResourceID resourceId,
			JobGraph jobGraph,
			HighAvailabilityServices highAvailabilityService,
			SlotPoolFactory slotPoolFactory,
			SchedulerFactory schedulerFactory,
			JobManagerSharedServices jobManagerSharedServices,
			HeartbeatServices heartbeatServices,
			JobManagerJobMetricGroupFactory jobMetricGroupFactory,
			OnCompletionActions jobCompletionActions,
			FatalErrorHandler fatalErrorHandler,
			ClassLoader userCodeLoader); CompletableFuture<Acknowledge> start(final JobMasterId newJobMasterId); CompletableFuture<Acknowledge> suspend(final Exception cause); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<Acknowledge> cancel(Time timeout); @Override CompletableFuture<Acknowledge> stop(Time timeout); @Override CompletableFuture<Acknowledge> rescaleJob(
			int newParallelism,
			RescalingBehaviour rescalingBehaviour,
			Time timeout); @Override CompletableFuture<Acknowledge> rescaleOperators(
			Collection<JobVertexID> operators,
			int newParallelism,
			RescalingBehaviour rescalingBehaviour,
			Time timeout); @Override CompletableFuture<Acknowledge> updateTaskExecutionState(
			final TaskExecutionState taskExecutionState); @Override CompletableFuture<SerializedInputSplit> requestNextInputSplit(
			final JobVertexID vertexID,
			final ExecutionAttemptID executionAttempt); @Override CompletableFuture<ExecutionState> requestPartitionState(
			final IntermediateDataSetID intermediateResultId,
			final ResultPartitionID resultPartitionId); @Override CompletableFuture<Acknowledge> scheduleOrUpdateConsumers(
			final ResultPartitionID partitionID,
			final Time timeout); @Override CompletableFuture<Acknowledge> disconnectTaskManager(final ResourceID resourceID, final Exception cause); @Override void acknowledgeCheckpoint(
			final JobID jobID,
			final ExecutionAttemptID executionAttemptID,
			final long checkpointId,
			final CheckpointMetrics checkpointMetrics,
			final TaskStateSnapshot checkpointState); @Override void declineCheckpoint(DeclineCheckpoint decline); @Override CompletableFuture<KvStateLocation> requestKvStateLocation(final JobID jobId, final String registrationName); @Override CompletableFuture<Acknowledge> notifyKvStateRegistered(
			final JobID jobId,
			final JobVertexID jobVertexId,
			final KeyGroupRange keyGroupRange,
			final String registrationName,
			final KvStateID kvStateId,
			final InetSocketAddress kvStateServerAddress); @Override CompletableFuture<Acknowledge> notifyKvStateUnregistered(
			JobID jobId,
			JobVertexID jobVertexId,
			KeyGroupRange keyGroupRange,
			String registrationName); @Override CompletableFuture<Collection<SlotOffer>> offerSlots(
			final ResourceID taskManagerId,
			final Collection<SlotOffer> slots,
			final Time timeout); @Override void failSlot(
			final ResourceID taskManagerId,
			final AllocationID allocationId,
			final Exception cause); @Override CompletableFuture<RegistrationResponse> registerTaskManager(
			final String taskManagerRpcAddress,
			final TaskManagerLocation taskManagerLocation,
			final Time timeout); @Override void disconnectResourceManager(
			final ResourceManagerId resourceManagerId,
			final Exception cause); @Override void heartbeatFromTaskManager(final ResourceID resourceID, AccumulatorReport accumulatorReport); @Override void heartbeatFromResourceManager(final ResourceID resourceID); @Override CompletableFuture<JobDetails> requestJobDetails(Time timeout); @Override CompletableFuture<JobStatus> requestJobStatus(Time timeout); @Override CompletableFuture<ArchivedExecutionGraph> requestJob(Time timeout); @Override CompletableFuture<String> triggerSavepoint(
			@Nullable final String targetDirectory,
			final boolean cancelJob,
			final Time timeout); @Override CompletableFuture<OperatorBackPressureStatsResponse> requestOperatorBackPressureStats(final JobVertexID jobVertexId); @Override void notifyAllocationFailure(AllocationID allocationID, Exception cause); @Override CompletableFuture<Object> updateGlobalAggregate(String aggregateName, Object aggregand, byte[] serializedAggregateFunction); @Override JobMasterGateway getGateway(); static final String JOB_MANAGER_NAME; static final String ARCHIVE_NAME; }