@Test public void testSetupApplicationMasterContainer() { Configuration cfg = new Configuration(); YarnClusterDescriptor clusterDescriptor = new YarnClusterDescriptor( cfg, yarnConfiguration, temporaryFolder.getRoot().getAbsolutePath(), yarnClient, true); final String java = "$JAVA_HOME/bin/java"; final String jvmmem = "-Xms424m -Xmx424m"; final String jvmOpts = "-Djvm"; final String jmJvmOpts = "-DjmJvm"; final String krb5 = "-Djava.security.krb5.conf=krb5.conf"; final String logfile = "-Dlog.file=\"" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/jobmanager.log\""; final String logback = "-Dlogback.configurationFile=file:" + FlinkYarnSessionCli.CONFIG_FILE_LOGBACK_NAME; final String log4j = "-Dlog4j.configuration=file:" + FlinkYarnSessionCli.CONFIG_FILE_LOG4J_NAME; final String mainClass = clusterDescriptor.getYarnSessionClusterEntrypoint(); final String args = ""; final String redirects = "1> " + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/jobmanager.out " + "2> " + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/jobmanager.err"; final int jobManagerMemory = 1024; try { assertEquals( java + " " + jvmmem + " " + " " + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, false, false, false, jobManagerMemory) .getCommands().get(0)); assertEquals( java + " " + jvmmem + " " + " " + krb5 + " " + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, false, false, true, jobManagerMemory) .getCommands().get(0)); assertEquals( java + " " + jvmmem + " " + " " + logfile + " " + logback + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, true, false, false, jobManagerMemory) .getCommands().get(0)); assertEquals( java + " " + jvmmem + " " + " " + krb5 + " " + logfile + " " + logback + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, true, false, true, jobManagerMemory) .getCommands().get(0)); assertEquals( java + " " + jvmmem + " " + " " + logfile + " " + log4j + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, false, true, false, jobManagerMemory) .getCommands().get(0)); assertEquals( java + " " + jvmmem + " " + " " + krb5 + " " + logfile + " " + log4j + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, false, true, true, jobManagerMemory) .getCommands().get(0)); assertEquals( java + " " + jvmmem + " " + " " + logfile + " " + logback + " " + log4j + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, true, true, false, jobManagerMemory) .getCommands().get(0)); assertEquals( java + " " + jvmmem + " " + " " + krb5 + " " + logfile + " " + logback + " " + log4j + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, true, true, true, jobManagerMemory) .getCommands().get(0)); cfg.setString(CoreOptions.FLINK_JVM_OPTIONS, jvmOpts); assertEquals( java + " " + jvmmem + " " + jvmOpts + " " + logfile + " " + logback + " " + log4j + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, true, true, false, jobManagerMemory) .getCommands().get(0)); assertEquals( java + " " + jvmmem + " " + jvmOpts + " " + krb5 + " " + logfile + " " + logback + " " + log4j + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, true, true, true, jobManagerMemory) .getCommands().get(0)); cfg.setString(CoreOptions.FLINK_JM_JVM_OPTIONS, jmJvmOpts); assertEquals( java + " " + jvmmem + " " + jvmOpts + " " + jmJvmOpts + " " + logfile + " " + logback + " " + log4j + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, true, true, false, jobManagerMemory) .getCommands().get(0)); assertEquals( java + " " + jvmmem + " " + jvmOpts + " " + jmJvmOpts + " " + krb5 + " " + logfile + " " + logback + " " + log4j + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, true, true, true, jobManagerMemory) .getCommands().get(0)); cfg.setString(ConfigConstants.YARN_CONTAINER_START_COMMAND_TEMPLATE, "%java% 1 %jvmmem% 2 %jvmopts% 3 %logging% 4 %class% 5 %args% 6 %redirects%"); assertEquals( java + " 1 " + jvmmem + " 2 " + jvmOpts + " " + jmJvmOpts + " " + krb5 + " 3 " + logfile + " " + logback + " " + log4j + " 4 " + mainClass + " 5 " + args + " 6 " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, true, true, true, jobManagerMemory) .getCommands().get(0)); cfg.setString(ConfigConstants.YARN_CONTAINER_START_COMMAND_TEMPLATE, "%java% %logging% %jvmopts% %jvmmem% %class% %args% %redirects%"); assertEquals( java + " " + logfile + " " + logback + " " + log4j + " " + jvmOpts + " " + jmJvmOpts + " " + krb5 + " " + jvmmem + " " + mainClass + " " + args + " " + redirects, clusterDescriptor .setupApplicationMasterContainer( mainClass, true, true, true, jobManagerMemory) .getCommands().get(0)); } finally { clusterDescriptor.close(); } }
@Override protected String getYarnSessionClusterEntrypoint() { return YarnSessionClusterEntrypoint.class.getName(); }
YarnClusterDescriptor extends AbstractYarnClusterDescriptor { @Override protected String getYarnSessionClusterEntrypoint() { return YarnSessionClusterEntrypoint.class.getName(); } }
YarnClusterDescriptor extends AbstractYarnClusterDescriptor { @Override protected String getYarnSessionClusterEntrypoint() { return YarnSessionClusterEntrypoint.class.getName(); } YarnClusterDescriptor(
			Configuration flinkConfiguration,
			YarnConfiguration yarnConfiguration,
			String configurationDirectory,
			YarnClient yarnClient,
			boolean sharedYarnClient); }
YarnClusterDescriptor extends AbstractYarnClusterDescriptor { @Override protected String getYarnSessionClusterEntrypoint() { return YarnSessionClusterEntrypoint.class.getName(); } YarnClusterDescriptor(
			Configuration flinkConfiguration,
			YarnConfiguration yarnConfiguration,
			String configurationDirectory,
			YarnClient yarnClient,
			boolean sharedYarnClient); @Override ClusterClient<ApplicationId> deployJobCluster(
		ClusterSpecification clusterSpecification,
		JobGraph jobGraph,
		boolean detached); }
YarnClusterDescriptor extends AbstractYarnClusterDescriptor { @Override protected String getYarnSessionClusterEntrypoint() { return YarnSessionClusterEntrypoint.class.getName(); } YarnClusterDescriptor(
			Configuration flinkConfiguration,
			YarnConfiguration yarnConfiguration,
			String configurationDirectory,
			YarnClient yarnClient,
			boolean sharedYarnClient); @Override ClusterClient<ApplicationId> deployJobCluster(
		ClusterSpecification clusterSpecification,
		JobGraph jobGraph,
		boolean detached);  }