@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);
		// wait for the start operation to complete
		startFuture.get(testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS);
		// define a leader and see that a registration happens
		rmLeaderRetrievalService.notifyListener(resourceManagerAddress, resourceManagerId.toUUID());
		// register job manager success will trigger monitor heartbeat target between jm and rm
		verify(resourceManagerGateway, timeout(testingTimeout.toMilliseconds())).registerJobManager(
			eq(jobMasterId),
			eq(jmResourceId),
			anyString(),
			eq(jobGraph.getJobID()),
			any(Time.class));
		// heartbeat timeout should trigger disconnect JobManager from ResourceManager
		verify(resourceManagerGateway, timeout(heartbeatTimeout * 50L)).disconnectJobManager(eq(jobGraph.getJobID()), any(TimeoutException.class));
		// check if a concurrent error occurred
		testingFatalErrorHandler.rethrowError();
	} finally {
		rpc.stopService();
	}
}