flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
/**
	 * Test the YARN Java API.
	 */
Test the YARN Java API.
Test the YARN Java API .
@Test
	public void testJavaAPI() throws Exception {
		final int waitTime = 15;
		LOG.info("Starting testJavaAPI()");
		String confDirPath = System.getenv(ConfigConstants.ENV_FLINK_CONF_DIR);
		Configuration configuration = GlobalConfiguration.loadConfiguration();
		AbstractYarnClusterDescriptor flinkYarnClient = new YarnClusterDescriptor(configuration, confDirPath);
		Assert.assertNotNull("unable to get yarn client", flinkYarnClient);
		flinkYarnClient.setLocalJarPath(new Path(flinkUberjar.getAbsolutePath()));
		flinkYarnClient.addShipFiles(Arrays.asList(flinkLibFolder.listFiles()));
		final ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
			.setMasterMemoryMB(768)
			.setTaskManagerMemoryMB(1024)
			.setNumberTaskManagers(1)
			.setSlotsPerTaskManager(1)
			.createClusterSpecification();
		ClusterClient yarnCluster = null;
		try {
			yarnCluster = flinkYarnClient.deploySessionCluster(clusterSpecification);
		} catch (Exception e) {
			LOG.warn("Failing test", e);
			Assert.fail("Error while deploying YARN cluster: " + e.getMessage());
		}
		GetClusterStatusResponse expectedStatus = new GetClusterStatusResponse(1, 1);
		for (int second = 0; second < waitTime * 2; second++) { 
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				LOG.warn("Interrupted", e);
			}
			GetClusterStatusResponse status = yarnCluster.getClusterStatus();
			if (status != null && status.equals(expectedStatus)) {
				LOG.info("ClusterClient reached status " + status);
				break; 
			}
			if (second > waitTime) {
				Assert.fail("The custer didn't start after " + waitTime + " seconds");
			}
		}
		Assert.assertNotNull(yarnCluster.getJobManagerAddress());
		Assert.assertNotNull(yarnCluster.getWebInterfaceURL());
		LOG.info("Shutting down cluster. All tests passed");
		yarnCluster.shutdown();
		LOG.info("Finished testJavaAPI()");
	}
@ Test public void testJavaAPI ( ) throws Exception { final int waitTime = 15 ; LOG . info ( "Starting testJavaAPI()" ) ; String confDirPath = System . getenv ( ConfigConstants . ENV_FLINK_CONF_DIR ) ; Configuration configuration = GlobalConfiguration . loadConfiguration ( ) ; AbstractYarnClusterDescriptor flinkYarnClient = new YarnClusterDescriptor ( configuration , confDirPath ) ; Assert . assertNotNull ( "unable to get yarn client" , flinkYarnClient ) ; flinkYarnClient . setLocalJarPath ( new Path ( flinkUberjar . getAbsolutePath ( ) ) ) ; flinkYarnClient . addShipFiles ( Arrays . asList ( flinkLibFolder . listFiles ( ) ) ) ; final ClusterSpecification clusterSpecification = new ClusterSpecification . ClusterSpecificationBuilder ( ) . setMasterMemoryMB ( 768 ) . setTaskManagerMemoryMB ( 1024 ) . setNumberTaskManagers ( 1 ) . setSlotsPerTaskManager ( 1 ) . createClusterSpecification ( ) ; ClusterClient yarnCluster = null ; try { yarnCluster = flinkYarnClient . deploySessionCluster ( clusterSpecification ) ; } catch ( Exception e ) { LOG . warn ( "Failing test" , e ) ; Assert . fail ( "Error while deploying YARN cluster: " + e . getMessage ( ) ) ; } GetClusterStatusResponse expectedStatus = new GetClusterStatusResponse ( 1 , 1 ) ; for ( int second = 0 ; second < waitTime * 2 ; second ++ ) { try { Thread . sleep ( 1000 ) ; } catch ( InterruptedException e ) { LOG . warn ( "Interrupted" , e ) ; } GetClusterStatusResponse status = yarnCluster . getClusterStatus ( ) ; if ( status != null && status . equals ( expectedStatus ) ) { LOG . info ( "ClusterClient reached status " + status ) ; break ; } if ( second > waitTime ) { Assert . fail ( "The custer didn't start after " + waitTime + " seconds" ) ; } } Assert . assertNotNull ( yarnCluster . getJobManagerAddress ( ) ) ; Assert . assertNotNull ( yarnCluster . getWebInterfaceURL ( ) ) ; LOG . info ( "Shutting down cluster. All tests passed" ) ; yarnCluster . shutdown ( ) ; LOG . info ( "Finished testJavaAPI()" ) ; }
Test the YARN Java API.
Test the YARN Java API .
// deploy // run "forever" // all good, cluster started // we waited for 15 seconds. cluster didn't come up correctly // use the cluster // shutdown cluster