JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway { @Override public void start() { throw new UnsupportedOperationException("Should never call start() without leader ID"); } JobMaster(
			RpcService rpcService,
			ResourceID resourceId,
			JobGraph jobGraph,
			Configuration configuration,
			HighAvailabilityServices highAvailabilityService,
			HeartbeatServices heartbeatServices,
			ScheduledExecutorService executor,
			BlobServer blobServer,
			BlobLibraryCacheManager libraryCacheManager,
			RestartStrategyFactory restartStrategyFactory,
			Time rpcAskTimeout,
			@Nullable JobManagerMetricGroup jobManagerMetricGroup,
			OnCompletionActions jobCompletionActions,
			FatalErrorHandler errorHandler,
			ClassLoader userCodeLoader); @Override void start(); CompletableFuture<Acknowledge> start(final JobMasterId newJobMasterId, final Time timeout); CompletableFuture<Acknowledge> suspend(final Throwable cause, final Time timeout); @Override void postStop(); @Override CompletableFuture<Acknowledge> cancel(Time timeout); @Override CompletableFuture<Acknowledge> stop(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 void 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(
			final JobID jobID,
			final ExecutionAttemptID executionAttemptID,
			final long checkpointID,
			final Throwable reason); @Override CompletableFuture<KvStateLocation> lookupKvStateLocation(final String registrationName); @Override void notifyKvStateRegistered(
			final JobVertexID jobVertexId,
			final KeyGroupRange keyGroupRange,
			final String registrationName,
			final KvStateID kvStateId,
			final InetSocketAddress kvStateServerAddress); @Override void notifyKvStateUnregistered(
			JobVertexID jobVertexId,
			KeyGroupRange keyGroupRange,
			String registrationName); @Override CompletableFuture<ClassloadingProps> requestClassloadingProps(); @Override CompletableFuture<Collection<SlotOffer>> offerSlots(
			final ResourceID taskManagerId,
			final Iterable<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); @Override void heartbeatFromResourceManager(final ResourceID resourceID); @Override CompletableFuture<JobDetails> requestJobDetails(Time timeout); @Override CompletableFuture<AccessExecutionGraph> requestArchivedExecutionGraph(Time timeout); @Override CompletableFuture<JobStatus> requestJobStatus(Time timeout); static final String JOB_MANAGER_NAME; static final String ARCHIVE_NAME; }
@Test public void testHeartbeatTimeoutWithResourceManager() throws Exception { final String resourceManagerAddress = "rm"; final String jobManagerAddress = "jm"; final ResourceManagerId resourceManagerId = ResourceManagerId.generate(); final JobMasterId jobMasterId = JobMasterId.generate(); final ResourceID rmResourceId = new ResourceID(resourceManagerAddress); final ResourceID jmResourceId = new ResourceID(jobManagerAddress); final JobGraph jobGraph = new JobGraph(); final TestingHighAvailabilityServices haServices = new TestingHighAvailabilityServices(); final TestingLeaderRetrievalService rmLeaderRetrievalService = new TestingLeaderRetrievalService( null, null); haServices.setResourceManagerLeaderRetriever(rmLeaderRetrievalService); haServices.setCheckpointRecoveryFactory(mock(CheckpointRecoveryFactory.class)); final long heartbeatInterval = 1L; final long heartbeatTimeout = 5L; final ScheduledExecutor scheduledExecutor = mock(ScheduledExecutor.class); final HeartbeatServices heartbeatServices = new TestingHeartbeatServices(heartbeatInterval, heartbeatTimeout, scheduledExecutor); final ResourceManagerGateway resourceManagerGateway = mock(ResourceManagerGateway.class); when(resourceManagerGateway.registerJobManager( any(JobMasterId.class), any(ResourceID.class), anyString(), any(JobID.class), any(Time.class) )).thenReturn(CompletableFuture.completedFuture(new JobMasterRegistrationSuccess( heartbeatInterval, resourceManagerId, rmResourceId))); final TestingRpcService rpc = new TestingRpcService(); rpc.registerGateway(resourceManagerAddress, resourceManagerGateway); final TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler(); Configuration configuration = new Configuration(); try (BlobServer blobServer = new BlobServer(configuration, new VoidBlobStore())) { blobServer.start(); final JobMaster jobMaster = new JobMaster( rpc, jmResourceId, jobGraph, configuration, haServices, heartbeatServices, Executors.newScheduledThreadPool(1), blobServer, new BlobLibraryCacheManager( blobServer, FlinkUserCodeClassLoaders.ResolveOrder.CHILD_FIRST, new String[0]), mock(RestartStrategyFactory.class), testingTimeout, null, mock(OnCompletionActions.class), testingFatalErrorHandler, FlinkUserCodeClassLoaders.parentFirst(new URL[0], JobMasterTest.class.getClassLoader())); CompletableFuture<Acknowledge> startFuture = jobMaster.start(jobMasterId, testingTimeout); startFuture.get(testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS); rmLeaderRetrievalService.notifyListener(resourceManagerAddress, resourceManagerId.toUUID()); verify(resourceManagerGateway, timeout(testingTimeout.toMilliseconds())).registerJobManager( eq(jobMasterId), eq(jmResourceId), anyString(), eq(jobGraph.getJobID()), any(Time.class)); verify(resourceManagerGateway, timeout(heartbeatTimeout * 50L)).disconnectJobManager(eq(jobGraph.getJobID()), any(TimeoutException.class)); testingFatalErrorHandler.rethrowError(); } finally { rpc.stopService(); } }