flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerJobMasterTest.java
/**
	 * Check and verify return RegistrationResponse.Decline when failed to start a job master Leader retrieval listener
	 */
Check and verify return RegistrationResponse.Decline when failed to start a job master Leader retrieval listener
Check and verify return RegistrationResponse . Decline when failed to start a job master Leader retrieval listener
@Test
	public void testRegisterJobMasterWithFailureLeaderListener() throws Exception {
		String jobMasterAddress = "/jobMasterAddress1";
		JobID jobID = mockJobMaster(jobMasterAddress);
		TestingLeaderElectionService resourceManagerLeaderElectionService = new TestingLeaderElectionService();
		TestingLeaderRetrievalService jobMasterLeaderRetrievalService = new TestingLeaderRetrievalService(
			"localhost",
			HighAvailabilityServices.DEFAULT_LEADER_ID);
		TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler();
		final ResourceManager<?> resourceManager = createAndStartResourceManager(
			resourceManagerLeaderElectionService,
			jobID,
			jobMasterLeaderRetrievalService,
			testingFatalErrorHandler);
		final ResourceManagerGateway rmGateway = resourceManager.getSelfGateway(ResourceManagerGateway.class);
		final ResourceID jmResourceId = new ResourceID(jobMasterAddress);
		JobID unknownJobIDToHAServices = new JobID();
		resourceManagerLeaderElectionService.isLeader(UUID.randomUUID()).get();
		CompletableFuture<RegistrationResponse> registrationFuture = rmGateway.registerJobManager(
			new JobMasterId(HighAvailabilityServices.DEFAULT_LEADER_ID),
			jmResourceId,
			jobMasterAddress,
			unknownJobIDToHAServices,
			timeout);
		try {
			registrationFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
		} catch (ExecutionException e) {
			assertTrue(ExceptionUtils.stripExecutionException(e) instanceof ResourceManagerException);
		}
	}
@ Test public void testRegisterJobMasterWithFailureLeaderListener ( ) throws Exception { String jobMasterAddress = "/jobMasterAddress1" ; JobID jobID = mockJobMaster ( jobMasterAddress ) ; TestingLeaderElectionService resourceManagerLeaderElectionService = new TestingLeaderElectionService ( ) ; TestingLeaderRetrievalService jobMasterLeaderRetrievalService = new TestingLeaderRetrievalService ( "localhost" , HighAvailabilityServices . DEFAULT_LEADER_ID ) ; TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler ( ) ; final ResourceManager < ? > resourceManager = createAndStartResourceManager ( resourceManagerLeaderElectionService , jobID , jobMasterLeaderRetrievalService , testingFatalErrorHandler ) ; final ResourceManagerGateway rmGateway = resourceManager . getSelfGateway ( ResourceManagerGateway . class ) ; final ResourceID jmResourceId = new ResourceID ( jobMasterAddress ) ; JobID unknownJobIDToHAServices = new JobID ( ) ; resourceManagerLeaderElectionService . isLeader ( UUID . randomUUID ( ) ) . get ( ) ; CompletableFuture < RegistrationResponse > registrationFuture = rmGateway . registerJobManager ( new JobMasterId ( HighAvailabilityServices . DEFAULT_LEADER_ID ) , jmResourceId , jobMasterAddress , unknownJobIDToHAServices , timeout ) ; try { registrationFuture . get ( timeout . toMilliseconds ( ) , TimeUnit . MILLISECONDS ) ; } catch ( ExecutionException e ) { assertTrue ( ExceptionUtils . stripExecutionException ( e ) instanceof ResourceManagerException ) ; } }
Check and verify return RegistrationResponse.Decline when failed to start a job master Leader retrieval listener
Check and verify return RegistrationResponse . Decline when failed to start a job master Leader retrieval listener
// wait until the leader election has been completed // this should fail because we try to register a job leader listener for an unknown job id