flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
/**
	 * Tests that the application master can be killed multiple times and that the surviving
	 * TaskManager successfully reconnects to the newly started JobManager.
	 * @throws Exception
	 */
Tests that the application master can be killed multiple times and that the surviving
TaskManager successfully reconnects to the newly started JobManager.
@throws Exception
Tests that the application master can be killed multiple times and that the surviving TaskManager successfully reconnects to the newly started JobManager . @throws Exception
@Test
	public void testMultipleAMKill() throws Exception {
		final int numberKillingAttempts = numberApplicationAttempts - 1;
		String confDirPath = System.getenv(ConfigConstants.ENV_FLINK_CONF_DIR);
		final Configuration configuration = GlobalConfiguration.loadConfiguration();
		TestingYarnClusterDescriptor flinkYarnClient = new TestingYarnClusterDescriptor(configuration, confDirPath);
		Assert.assertNotNull("unable to get yarn client", flinkYarnClient);
		flinkYarnClient.setLocalJarPath(new Path(flinkUberjar.getAbsolutePath()));
		flinkYarnClient.addShipFiles(Arrays.asList(flinkLibFolder.listFiles()));
		String fsStateHandlePath = temp.getRoot().getPath();
		File configDirectory = new File(confDirPath);
		GlobalConfiguration.loadConfiguration(configDirectory.getAbsolutePath());
		flinkYarnClient.setDynamicPropertiesEncoded("recovery.mode=zookeeper@@recovery.zookeeper.quorum=" +
			zkServer.getConnectString() + "@@yarn.application-attempts=" + numberApplicationAttempts +
			"@@" + CoreOptions.STATE_BACKEND + "=FILESYSTEM" +
			"@@" + FsStateBackendFactory.CHECKPOINT_DIRECTORY_URI_CONF_KEY + "=" + fsStateHandlePath + "/checkpoints" +
			"@@" + HighAvailabilityOptions.HA_STORAGE_PATH.key() + "=" + fsStateHandlePath + "/recovery");
		ClusterClient yarnCluster = null;
		final FiniteDuration timeout = new FiniteDuration(2, TimeUnit.MINUTES);
		HighAvailabilityServices highAvailabilityServices = null;
		final ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
			.setMasterMemoryMB(768)
			.setTaskManagerMemoryMB(1024)
			.setNumberTaskManagers(1)
			.setSlotsPerTaskManager(1)
			.createClusterSpecification();
		try {
			yarnCluster = flinkYarnClient.deploySessionCluster(clusterSpecification);
			final ClusterClient finalYarnCluster = yarnCluster;
			highAvailabilityServices = HighAvailabilityServicesUtils.createHighAvailabilityServices(
				yarnCluster.getFlinkConfiguration(),
				Executors.directExecutor(),
				HighAvailabilityServicesUtils.AddressResolution.TRY_ADDRESS_RESOLUTION);
			final HighAvailabilityServices finalHighAvailabilityServices = highAvailabilityServices;
			new JavaTestKit(actorSystem) {{
				for (int attempt = 0; attempt < numberKillingAttempts; attempt++) {
					new Within(timeout) {
						@Override
						protected void run() {
							try {
								ActorGateway gateway = LeaderRetrievalUtils.retrieveLeaderGateway(
									finalHighAvailabilityServices.getJobManagerLeaderRetriever(HighAvailabilityServices.DEFAULT_JOB_ID),
									actorSystem,
									timeout);
								ActorGateway selfGateway = new AkkaActorGateway(getRef(), gateway.leaderSessionID());
								gateway.tell(new TestingJobManagerMessages.NotifyWhenAtLeastNumTaskManagerAreRegistered(1), selfGateway);
								expectMsgEquals(Acknowledge.get());
								gateway.tell(PoisonPill.getInstance());
							} catch (Exception e) {
								throw new AssertionError("Could not complete test.", e);
							}
						}
					};
				}
				new Within(timeout) {
					@Override
					protected void run() {
						try {
							ActorGateway gateway = LeaderRetrievalUtils.retrieveLeaderGateway(
								finalHighAvailabilityServices.getJobManagerLeaderRetriever(HighAvailabilityServices.DEFAULT_JOB_ID),
								actorSystem,
								timeout);
							ActorGateway selfGateway = new AkkaActorGateway(getRef(), gateway.leaderSessionID());
							gateway.tell(new TestingJobManagerMessages.NotifyWhenAtLeastNumTaskManagerAreRegistered(1), selfGateway);
							expectMsgEquals(Acknowledge.get());
						} catch (Exception e) {
							throw new AssertionError("Could not complete test.", e);
						}
					}
				};
			}};
		} finally {
			if (yarnCluster != null) {
				yarnCluster.shutdown();
			}
			if (highAvailabilityServices != null) {
				highAvailabilityServices.closeAndCleanupAllData();
			}
		}
	}
@ Test public void testMultipleAMKill ( ) throws Exception { final int numberKillingAttempts = numberApplicationAttempts - 1 ; String confDirPath = System . getenv ( ConfigConstants . ENV_FLINK_CONF_DIR ) ; final Configuration configuration = GlobalConfiguration . loadConfiguration ( ) ; TestingYarnClusterDescriptor flinkYarnClient = new TestingYarnClusterDescriptor ( configuration , confDirPath ) ; Assert . assertNotNull ( "unable to get yarn client" , flinkYarnClient ) ; flinkYarnClient . setLocalJarPath ( new Path ( flinkUberjar . getAbsolutePath ( ) ) ) ; flinkYarnClient . addShipFiles ( Arrays . asList ( flinkLibFolder . listFiles ( ) ) ) ; String fsStateHandlePath = temp . getRoot ( ) . getPath ( ) ; File configDirectory = new File ( confDirPath ) ; GlobalConfiguration . loadConfiguration ( configDirectory . getAbsolutePath ( ) ) ; flinkYarnClient . setDynamicPropertiesEncoded ( "recovery.mode=zookeeper@@recovery.zookeeper.quorum=" + zkServer . getConnectString ( ) + "@@yarn.application-attempts=" + numberApplicationAttempts + "@@" + CoreOptions . STATE_BACKEND + "=FILESYSTEM" + "@@" + FsStateBackendFactory . CHECKPOINT_DIRECTORY_URI_CONF_KEY + "=" + fsStateHandlePath + "/checkpoints" + "@@" + HighAvailabilityOptions . HA_STORAGE_PATH . key ( ) + "=" + fsStateHandlePath + "/recovery" ) ; ClusterClient yarnCluster = null ; final FiniteDuration timeout = new FiniteDuration ( 2 , TimeUnit . MINUTES ) ; HighAvailabilityServices highAvailabilityServices = null ; final ClusterSpecification clusterSpecification = new ClusterSpecification . ClusterSpecificationBuilder ( ) . setMasterMemoryMB ( 768 ) . setTaskManagerMemoryMB ( 1024 ) . setNumberTaskManagers ( 1 ) . setSlotsPerTaskManager ( 1 ) . createClusterSpecification ( ) ; try { yarnCluster = flinkYarnClient . deploySessionCluster ( clusterSpecification ) ; final ClusterClient finalYarnCluster = yarnCluster ; highAvailabilityServices = HighAvailabilityServicesUtils . createHighAvailabilityServices ( yarnCluster . getFlinkConfiguration ( ) , Executors . directExecutor ( ) , HighAvailabilityServicesUtils . AddressResolution . TRY_ADDRESS_RESOLUTION ) ; final HighAvailabilityServices finalHighAvailabilityServices = highAvailabilityServices ; new JavaTestKit ( actorSystem ) { { for ( int attempt = 0 ; attempt < numberKillingAttempts ; attempt ++ ) { new Within ( timeout ) { @ Override protected void run ( ) { try { ActorGateway gateway = LeaderRetrievalUtils . retrieveLeaderGateway ( finalHighAvailabilityServices . getJobManagerLeaderRetriever ( HighAvailabilityServices . DEFAULT_JOB_ID ) , actorSystem , timeout ) ; ActorGateway selfGateway = new AkkaActorGateway ( getRef ( ) , gateway . leaderSessionID ( ) ) ; gateway . tell ( new TestingJobManagerMessages . NotifyWhenAtLeastNumTaskManagerAreRegistered ( 1 ) , selfGateway ) ; expectMsgEquals ( Acknowledge . get ( ) ) ; gateway . tell ( PoisonPill . getInstance ( ) ) ; } catch ( Exception e ) { throw new AssertionError ( "Could not complete test." , e ) ; } } } ; } new Within ( timeout ) { @ Override protected void run ( ) { try { ActorGateway gateway = LeaderRetrievalUtils . retrieveLeaderGateway ( finalHighAvailabilityServices . getJobManagerLeaderRetriever ( HighAvailabilityServices . DEFAULT_JOB_ID ) , actorSystem , timeout ) ; ActorGateway selfGateway = new AkkaActorGateway ( getRef ( ) , gateway . leaderSessionID ( ) ) ; gateway . tell ( new TestingJobManagerMessages . NotifyWhenAtLeastNumTaskManagerAreRegistered ( 1 ) , selfGateway ) ; expectMsgEquals ( Acknowledge . get ( ) ) ; } catch ( Exception e ) { throw new AssertionError ( "Could not complete test." , e ) ; } } } ; } } ; } finally { if ( yarnCluster != null ) { yarnCluster . shutdown ( ) ; } if ( highAvailabilityServices != null ) { highAvailabilityServices . closeAndCleanupAllData ( ) ; } } }
Tests that the application master can be killed multiple times and that the surviving
 TaskManager successfully reconnects to the newly started JobManager.
Tests that the application master can be killed multiple times and that the surviving TaskManager successfully reconnects to the newly started JobManager .
// load the configuration