@Test public void testHeartbeatTimeoutWithResourceManager() throws Exception { final String resourceManagerAddress = "rm"; final ResourceManagerId resourceManagerId = ResourceManagerId.generate(); final ResourceID rmResourceId = new ResourceID(resourceManagerAddress); final TestingResourceManagerGateway resourceManagerGateway = new TestingResourceManagerGateway( resourceManagerId, rmResourceId, resourceManagerAddress, "localhost"); final CompletableFuture<Tuple3<JobMasterId, ResourceID, JobID>> jobManagerRegistrationFuture = new CompletableFuture<>(); final CompletableFuture<JobID> disconnectedJobManagerFuture = new CompletableFuture<>(); final CountDownLatch registrationAttempts = new CountDownLatch(2); resourceManagerGateway.setRegisterJobManagerConsumer(tuple -> { jobManagerRegistrationFuture.complete( Tuple3.of( tuple.f0, tuple.f1, tuple.f3)); registrationAttempts.countDown(); }); resourceManagerGateway.setDisconnectJobManagerConsumer(tuple -> disconnectedJobManagerFuture.complete(tuple.f0)); rpcService.registerGateway(resourceManagerAddress, resourceManagerGateway); final JobManagerSharedServices jobManagerSharedServices = new TestingJobManagerSharedServicesBuilder().build(); final JobMaster jobMaster = createJobMaster( configuration, jobGraph, haServices, jobManagerSharedServices); CompletableFuture<Acknowledge> startFuture = jobMaster.start(jobMasterId); try { startFuture.get(testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS); rmLeaderRetrievalService.notifyListener(resourceManagerAddress, resourceManagerId.toUUID()); final Tuple3<JobMasterId, ResourceID, JobID> registrationInformation = jobManagerRegistrationFuture.get( testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS); assertThat(registrationInformation.f0, Matchers.equalTo(jobMasterId)); assertThat(registrationInformation.f1, Matchers.equalTo(jmResourceId)); assertThat(registrationInformation.f2, Matchers.equalTo(jobGraph.getJobID())); final JobID disconnectedJobManager = disconnectedJobManagerFuture.get(testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS); assertThat(disconnectedJobManager, Matchers.equalTo(jobGraph.getJobID())); registrationAttempts.await(); } finally { jobManagerSharedServices.shutdown(); RpcUtils.terminateRpcEndpoint(jobMaster, testingTimeout); } }
public CompletableFuture<Acknowledge> start(final JobMasterId newJobMasterId) throws Exception { start(); return callAsyncWithoutFencing(() -> startJobExecution(newJobMasterId), RpcUtils.INF_TIMEOUT); }
JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway, JobMasterService { public CompletableFuture<Acknowledge> start(final JobMasterId newJobMasterId) throws Exception { start(); return callAsyncWithoutFencing(() -> startJobExecution(newJobMasterId), RpcUtils.INF_TIMEOUT); } }
JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway, JobMasterService { public CompletableFuture<Acknowledge> start(final JobMasterId newJobMasterId) throws Exception { start(); return callAsyncWithoutFencing(() -> startJobExecution(newJobMasterId), RpcUtils.INF_TIMEOUT); } 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 { public CompletableFuture<Acknowledge> start(final JobMasterId newJobMasterId) throws Exception { start(); return callAsyncWithoutFencing(() -> startJobExecution(newJobMasterId), RpcUtils.INF_TIMEOUT); } 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 { public CompletableFuture<Acknowledge> start(final JobMasterId newJobMasterId) throws Exception { start(); return callAsyncWithoutFencing(() -> startJobExecution(newJobMasterId), RpcUtils.INF_TIMEOUT); } 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; }