flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerJobMasterTest.java
/**
	 * Test receive registration with unmatched leadershipId from job master
	 */
Test receive registration with unmatched leadershipId from job master
Test receive registration with unmatched leadershipId from job master
@Test
	public void testRegisterJobMasterWithUnmatchedLeaderSessionId1() throws Exception {
		String jobMasterAddress = "/jobMasterAddress1";
		JobID jobID = mockJobMaster(jobMasterAddress);
		JobMasterId jobMasterId = JobMasterId.generate();
		final ResourceID jmResourceId = new ResourceID(jobMasterAddress);
		TestingLeaderRetrievalService jobMasterLeaderRetrievalService = new TestingLeaderRetrievalService(jobMasterAddress, jobMasterId.toUUID());
		TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler();
		final ResourceManager<?> resourceManager = createAndStartResourceManager(mock(LeaderElectionService.class), jobID, jobMasterLeaderRetrievalService, testingFatalErrorHandler);
		final ResourceManagerGateway wronglyFencedGateway = rpcService.connect(resourceManager.getAddress(), ResourceManagerId.generate(), ResourceManagerGateway.class)
			.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
		CompletableFuture<RegistrationResponse> unMatchedLeaderFuture = wronglyFencedGateway.registerJobManager(
			jobMasterId,
			jmResourceId,
			jobMasterAddress,
			jobID,
			timeout);
		try {
			unMatchedLeaderFuture.get(5L, TimeUnit.SECONDS);
			fail("Should fail because we are using the wrong fencing token.");
		} catch (ExecutionException e) {
			assertTrue(ExceptionUtils.stripExecutionException(e) instanceof FencingTokenException);
		}
		if (testingFatalErrorHandler.hasExceptionOccurred()) {
			testingFatalErrorHandler.rethrowError();
		}
	}
@ Test public void testRegisterJobMasterWithUnmatchedLeaderSessionId1 ( ) throws Exception { String jobMasterAddress = "/jobMasterAddress1" ; JobID jobID = mockJobMaster ( jobMasterAddress ) ; JobMasterId jobMasterId = JobMasterId . generate ( ) ; final ResourceID jmResourceId = new ResourceID ( jobMasterAddress ) ; TestingLeaderRetrievalService jobMasterLeaderRetrievalService = new TestingLeaderRetrievalService ( jobMasterAddress , jobMasterId . toUUID ( ) ) ; TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler ( ) ; final ResourceManager < ? > resourceManager = createAndStartResourceManager ( mock ( LeaderElectionService . class ) , jobID , jobMasterLeaderRetrievalService , testingFatalErrorHandler ) ; final ResourceManagerGateway wronglyFencedGateway = rpcService . connect ( resourceManager . getAddress ( ) , ResourceManagerId . generate ( ) , ResourceManagerGateway . class ) . get ( timeout . toMilliseconds ( ) , TimeUnit . MILLISECONDS ) ; CompletableFuture < RegistrationResponse > unMatchedLeaderFuture = wronglyFencedGateway . registerJobManager ( jobMasterId , jmResourceId , jobMasterAddress , jobID , timeout ) ; try { unMatchedLeaderFuture . get ( 5L , TimeUnit . SECONDS ) ; fail ( "Should fail because we are using the wrong fencing token." ) ; } catch ( ExecutionException e ) { assertTrue ( ExceptionUtils . stripExecutionException ( e ) instanceof FencingTokenException ) ; } if ( testingFatalErrorHandler . hasExceptionOccurred ( ) ) { testingFatalErrorHandler . rethrowError ( ) ; } }
Test receive registration with unmatched leadershipId from job master
Test receive registration with unmatched leadershipId from job master
// test throw exception when receive a registration from job master which takes unmatched leaderSessionId