flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-clients/src/test/java/org/apache/flink/client/program/ClientConnectionTest.java
/**
	 * FLINK-6629
	 *
	 * <p>Tests that the {@link HighAvailabilityServices} are respected when initializing the ClusterClient's
	 * {@link ActorSystem} and retrieving the leading JobManager.
	 */
FLINK-6629
Tests that the  HighAvailabilityServices are respected when initializing the ClusterClient's
ActorSystem and retrieving the leading JobManager.
FLINK - 6629 Tests that the HighAvailabilityServices are respected when initializing the ClusterClient ' s ActorSystem and retrieving the leading JobManager .
@Test
	public void testJobManagerRetrievalWithHAServices() throws Exception {
		final Configuration configuration = new Configuration();
		final TestingHighAvailabilityServices highAvailabilityServices = new TestingHighAvailabilityServices();
		final ActorSystem actorSystem = AkkaUtils.createDefaultActorSystem();
		ActorRef actorRef = null;
		final UUID leaderId = UUID.randomUUID();
		try {
			actorRef = actorSystem.actorOf(
				Props.create(
					JobClientActorTest.PlainActor.class,
					leaderId));
			final String expectedAddress = AkkaUtils.getAkkaURL(actorSystem, actorRef);
			final TestingLeaderRetrievalService testingLeaderRetrievalService = new TestingLeaderRetrievalService(expectedAddress, leaderId);
			highAvailabilityServices.setJobMasterLeaderRetriever(HighAvailabilityServices.DEFAULT_JOB_ID, testingLeaderRetrievalService);
			ClusterClient client = new StandaloneClusterClient(configuration, highAvailabilityServices);
			ActorGateway gateway = client.getJobManagerGateway();
			assertEquals(expectedAddress, gateway.path());
			assertEquals(leaderId, gateway.leaderSessionID());
		} finally {
			if (actorRef != null) {
				TestingUtils.stopActorGracefully(actorRef);
			}
			actorSystem.shutdown();
		}
	}
@ Test public void testJobManagerRetrievalWithHAServices ( ) throws Exception { final Configuration configuration = new Configuration ( ) ; final TestingHighAvailabilityServices highAvailabilityServices = new TestingHighAvailabilityServices ( ) ; final ActorSystem actorSystem = AkkaUtils . createDefaultActorSystem ( ) ; ActorRef actorRef = null ; final UUID leaderId = UUID . randomUUID ( ) ; try { actorRef = actorSystem . actorOf ( Props . create ( JobClientActorTest . PlainActor . class , leaderId ) ) ; final String expectedAddress = AkkaUtils . getAkkaURL ( actorSystem , actorRef ) ; final TestingLeaderRetrievalService testingLeaderRetrievalService = new TestingLeaderRetrievalService ( expectedAddress , leaderId ) ; highAvailabilityServices . setJobMasterLeaderRetriever ( HighAvailabilityServices . DEFAULT_JOB_ID , testingLeaderRetrievalService ) ; ClusterClient client = new StandaloneClusterClient ( configuration , highAvailabilityServices ) ; ActorGateway gateway = client . getJobManagerGateway ( ) ; assertEquals ( expectedAddress , gateway . path ( ) ) ; assertEquals ( leaderId , gateway . leaderSessionID ( ) ) ; } finally { if ( actorRef != null ) { TestingUtils . stopActorGracefully ( actorRef ) ; } actorSystem . shutdown ( ) ; } }
FLINK-6629
 <p>Tests that the {@link HighAvailabilityServices} are respected when initializing the ClusterClient's
 {@link ActorSystem} and retrieving the leading JobManager.
FLINK - 6629 <p > Tests that the { @link HighAvailabilityServices } are respected when initializing the ClusterClient ' s { @link ActorSystem } and retrieving the leading JobManager .