flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/retriever/impl/AkkaJobManagerRetrieverTest.java
/**
	 * Tests that we can retrieve the current leading job manager.
	 */
Tests that we can retrieve the current leading job manager.
Tests that we can retrieve the current leading job manager .
@Test
	public void testAkkaJobManagerRetrieval() throws Exception {
		AkkaJobManagerRetriever akkaJobManagerRetriever = new AkkaJobManagerRetriever(actorSystem, timeout, 0, Time.milliseconds(0L));
		TestingLeaderRetrievalService testingLeaderRetrievalService = new TestingLeaderRetrievalService();
		CompletableFuture<JobManagerGateway> gatewayFuture = akkaJobManagerRetriever.getFuture();
		final UUID leaderSessionId = UUID.randomUUID();
		ActorRef actorRef = null;
		try {
			actorRef = actorSystem.actorOf(
				Props.create(JobClientActorTest.PlainActor.class, leaderSessionId));
			final String address = actorRef.path().toString();
			testingLeaderRetrievalService.start(akkaJobManagerRetriever);
			assertFalse(gatewayFuture.isDone());
			testingLeaderRetrievalService.notifyListener(address, leaderSessionId);
			JobManagerGateway jobManagerGateway = gatewayFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
			assertEquals(address, jobManagerGateway.getAddress());
		} finally {
			testingLeaderRetrievalService.stop();
			if (actorRef != null) {
				TestingUtils.stopActorGracefully(actorRef);
			}
		}
	}
@ Test public void testAkkaJobManagerRetrieval ( ) throws Exception { AkkaJobManagerRetriever akkaJobManagerRetriever = new AkkaJobManagerRetriever ( actorSystem , timeout , 0 , Time . milliseconds ( 0L ) ) ; TestingLeaderRetrievalService testingLeaderRetrievalService = new TestingLeaderRetrievalService ( ) ; CompletableFuture < JobManagerGateway > gatewayFuture = akkaJobManagerRetriever . getFuture ( ) ; final UUID leaderSessionId = UUID . randomUUID ( ) ; ActorRef actorRef = null ; try { actorRef = actorSystem . actorOf ( Props . create ( JobClientActorTest . PlainActor . class , leaderSessionId ) ) ; final String address = actorRef . path ( ) . toString ( ) ; testingLeaderRetrievalService . start ( akkaJobManagerRetriever ) ; assertFalse ( gatewayFuture . isDone ( ) ) ; testingLeaderRetrievalService . notifyListener ( address , leaderSessionId ) ; JobManagerGateway jobManagerGateway = gatewayFuture . get ( timeout . toMilliseconds ( ) , TimeUnit . MILLISECONDS ) ; assertEquals ( address , jobManagerGateway . getAddress ( ) ) ; } finally { testingLeaderRetrievalService . stop ( ) ; if ( actorRef != null ) { TestingUtils . stopActorGracefully ( actorRef ) ; } } }
Tests that we can retrieve the current leading job manager.
Tests that we can retrieve the current leading job manager .
// check that the gateway future has not been completed since there is no leader yet