@Test public void testGetTaskManagerShellCommand() { final Configuration cfg = new Configuration(); final ContaineredTaskManagerParameters containeredParams = new ContaineredTaskManagerParameters(1024, 768, 256, 4, new HashMap<String, String>()); final String java = "$JAVA_HOME/bin/java"; final String jvmmem = "-Xms768m -Xmx768m -XX:MaxDirectMemorySize=256m"; final String jvmOpts = "-Djvm"; final String tmJvmOpts = "-DtmJvm"; final String logfile = "-Dlog.file=./logs/taskmanager.log"; final String logback = "-Dlogback.configurationFile=file:./conf/logback.xml"; final String log4j = "-Dlog4j.configuration=file:./conf/log4j.properties"; final String mainClass = "org.apache.flink.runtime.clusterframework.BootstrapToolsTest"; final String args = "--configDir ./conf"; final String redirects = "1> ./logs/taskmanager.out 2> ./logs/taskmanager.err"; assertEquals( java + " " + jvmmem + " " + " " + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", false, false, false, this.getClass())); final String krb5 = "-Djava.security.krb5.conf=krb5.conf"; assertEquals( java + " " + jvmmem + " " + " " + krb5 + " " + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", false, false, true, this.getClass())); assertEquals( java + " " + jvmmem + " " + " " + logfile + " " + logback + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", true, false, false, this.getClass())); assertEquals( java + " " + jvmmem + " " + " " + krb5 + " " + logfile + " " + logback + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", true, false, true, this.getClass())); assertEquals( java + " " + jvmmem + " " + " " + logfile + " " + log4j + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", false, true, false, this.getClass())); assertEquals( java + " " + jvmmem + " " + " " + krb5 + " " + logfile + " " + log4j + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", false, true, true, this.getClass())); assertEquals( java + " " + jvmmem + " " + " " + logfile + " " + logback + " " + log4j + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", true, true, false, this.getClass())); assertEquals( java + " " + jvmmem + " " + " " + krb5 + " " + logfile + " " + logback + " " + log4j + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", true, true, true, this.getClass())); cfg.setString(CoreOptions.FLINK_JVM_OPTIONS, jvmOpts); assertEquals( java + " " + jvmmem + " " + jvmOpts + " " + logfile + " " + logback + " " + log4j + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", true, true, false, this.getClass())); assertEquals( java + " " + jvmmem + " " + jvmOpts + " " + krb5 + " " + logfile + " " + logback + " " + log4j + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", true, true, true, this.getClass())); cfg.setString(CoreOptions.FLINK_TM_JVM_OPTIONS, tmJvmOpts); assertEquals( java + " " + jvmmem + " " + jvmOpts + " " + tmJvmOpts + " " + logfile + " " + logback + " " + log4j + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", true, true, false, this.getClass())); assertEquals( java + " " + jvmmem + " " + jvmOpts + " " + tmJvmOpts + " " + krb5 + " " + logfile + " " + logback + " " + log4j + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", true, true, true, this.getClass())); 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 + " " + tmJvmOpts + " " + krb5 + " 3 " + logfile + " " + logback + " " + log4j + " 4 " + mainClass + " 5 " + args + " 6 " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", true, true, true, this.getClass())); cfg.setString(ConfigConstants.YARN_CONTAINER_START_COMMAND_TEMPLATE, "%java% %logging% %jvmopts% %jvmmem% %class% %args% %redirects%"); assertEquals( java + " " + logfile + " " + logback + " " + log4j + " " + jvmOpts + " " + tmJvmOpts + " " + krb5 + " " + jvmmem + " " + mainClass + " " + args + " " + redirects, BootstrapTools .getTaskManagerShellCommand(cfg, containeredParams, "./conf", "./logs", true, true, true, this.getClass())); }
public static String getTaskManagerShellCommand( Configuration flinkConfig, ContaineredTaskManagerParameters tmParams, String configDirectory, String logDirectory, boolean hasLogback, boolean hasLog4j, boolean hasKrb5, Class<?> mainClass) { final Map<String, String> startCommandValues = new HashMap<>(); startCommandValues.put("java", "$JAVA_HOME/bin/java"); ArrayList<String> params = new ArrayList<>(); params.add(String.format("-Xms%dm", tmParams.taskManagerHeapSizeMB())); params.add(String.format("-Xmx%dm", tmParams.taskManagerHeapSizeMB())); if (tmParams.taskManagerDirectMemoryLimitMB() >= 0) { params.add(String.format("-XX:MaxDirectMemorySize=%dm", tmParams.taskManagerDirectMemoryLimitMB())); } startCommandValues.put("jvmmem", StringUtils.join(params, ' ')); String javaOpts = flinkConfig.getString(CoreOptions.FLINK_JVM_OPTIONS); if (flinkConfig.getString(CoreOptions.FLINK_TM_JVM_OPTIONS).length() > 0) { javaOpts += " " + flinkConfig.getString(CoreOptions.FLINK_TM_JVM_OPTIONS); } if (hasKrb5) { javaOpts += " -Djava.security.krb5.conf=krb5.conf"; } startCommandValues.put("jvmopts", javaOpts); String logging = ""; if (hasLogback || hasLog4j) { logging = "-Dlog.file=" + logDirectory + "/taskmanager.log"; if (hasLogback) { logging += " -Dlogback.configurationFile=file:" + configDirectory + "/logback.xml"; } if (hasLog4j) { logging += " -Dlog4j.configuration=file:" + configDirectory + "/log4j.properties"; } } startCommandValues.put("logging", logging); startCommandValues.put("class", mainClass.getName()); startCommandValues.put("redirects", "1> " + logDirectory + "/taskmanager.out " + "2> " + logDirectory + "/taskmanager.err"); startCommandValues.put("args", "--configDir " + configDirectory); final String commandTemplate = flinkConfig .getString(ConfigConstants.YARN_CONTAINER_START_COMMAND_TEMPLATE, ConfigConstants.DEFAULT_YARN_CONTAINER_START_COMMAND_TEMPLATE); String startCommand = getStartCommand(commandTemplate, startCommandValues); LOG.debug("TaskManager start command: " + startCommand); return startCommand; }
BootstrapTools { public static String getTaskManagerShellCommand( Configuration flinkConfig, ContaineredTaskManagerParameters tmParams, String configDirectory, String logDirectory, boolean hasLogback, boolean hasLog4j, boolean hasKrb5, Class<?> mainClass) { final Map<String, String> startCommandValues = new HashMap<>(); startCommandValues.put("java", "$JAVA_HOME/bin/java"); ArrayList<String> params = new ArrayList<>(); params.add(String.format("-Xms%dm", tmParams.taskManagerHeapSizeMB())); params.add(String.format("-Xmx%dm", tmParams.taskManagerHeapSizeMB())); if (tmParams.taskManagerDirectMemoryLimitMB() >= 0) { params.add(String.format("-XX:MaxDirectMemorySize=%dm", tmParams.taskManagerDirectMemoryLimitMB())); } startCommandValues.put("jvmmem", StringUtils.join(params, ' ')); String javaOpts = flinkConfig.getString(CoreOptions.FLINK_JVM_OPTIONS); if (flinkConfig.getString(CoreOptions.FLINK_TM_JVM_OPTIONS).length() > 0) { javaOpts += " " + flinkConfig.getString(CoreOptions.FLINK_TM_JVM_OPTIONS); } if (hasKrb5) { javaOpts += " -Djava.security.krb5.conf=krb5.conf"; } startCommandValues.put("jvmopts", javaOpts); String logging = ""; if (hasLogback || hasLog4j) { logging = "-Dlog.file=" + logDirectory + "/taskmanager.log"; if (hasLogback) { logging += " -Dlogback.configurationFile=file:" + configDirectory + "/logback.xml"; } if (hasLog4j) { logging += " -Dlog4j.configuration=file:" + configDirectory + "/log4j.properties"; } } startCommandValues.put("logging", logging); startCommandValues.put("class", mainClass.getName()); startCommandValues.put("redirects", "1> " + logDirectory + "/taskmanager.out " + "2> " + logDirectory + "/taskmanager.err"); startCommandValues.put("args", "--configDir " + configDirectory); final String commandTemplate = flinkConfig .getString(ConfigConstants.YARN_CONTAINER_START_COMMAND_TEMPLATE, ConfigConstants.DEFAULT_YARN_CONTAINER_START_COMMAND_TEMPLATE); String startCommand = getStartCommand(commandTemplate, startCommandValues); LOG.debug("TaskManager start command: " + startCommand); return startCommand; } }
BootstrapTools { public static String getTaskManagerShellCommand( Configuration flinkConfig, ContaineredTaskManagerParameters tmParams, String configDirectory, String logDirectory, boolean hasLogback, boolean hasLog4j, boolean hasKrb5, Class<?> mainClass) { final Map<String, String> startCommandValues = new HashMap<>(); startCommandValues.put("java", "$JAVA_HOME/bin/java"); ArrayList<String> params = new ArrayList<>(); params.add(String.format("-Xms%dm", tmParams.taskManagerHeapSizeMB())); params.add(String.format("-Xmx%dm", tmParams.taskManagerHeapSizeMB())); if (tmParams.taskManagerDirectMemoryLimitMB() >= 0) { params.add(String.format("-XX:MaxDirectMemorySize=%dm", tmParams.taskManagerDirectMemoryLimitMB())); } startCommandValues.put("jvmmem", StringUtils.join(params, ' ')); String javaOpts = flinkConfig.getString(CoreOptions.FLINK_JVM_OPTIONS); if (flinkConfig.getString(CoreOptions.FLINK_TM_JVM_OPTIONS).length() > 0) { javaOpts += " " + flinkConfig.getString(CoreOptions.FLINK_TM_JVM_OPTIONS); } if (hasKrb5) { javaOpts += " -Djava.security.krb5.conf=krb5.conf"; } startCommandValues.put("jvmopts", javaOpts); String logging = ""; if (hasLogback || hasLog4j) { logging = "-Dlog.file=" + logDirectory + "/taskmanager.log"; if (hasLogback) { logging += " -Dlogback.configurationFile=file:" + configDirectory + "/logback.xml"; } if (hasLog4j) { logging += " -Dlog4j.configuration=file:" + configDirectory + "/log4j.properties"; } } startCommandValues.put("logging", logging); startCommandValues.put("class", mainClass.getName()); startCommandValues.put("redirects", "1> " + logDirectory + "/taskmanager.out " + "2> " + logDirectory + "/taskmanager.err"); startCommandValues.put("args", "--configDir " + configDirectory); final String commandTemplate = flinkConfig .getString(ConfigConstants.YARN_CONTAINER_START_COMMAND_TEMPLATE, ConfigConstants.DEFAULT_YARN_CONTAINER_START_COMMAND_TEMPLATE); String startCommand = getStartCommand(commandTemplate, startCommandValues); LOG.debug("TaskManager start command: " + startCommand); return startCommand; } private  BootstrapTools(); }
BootstrapTools { public static String getTaskManagerShellCommand( Configuration flinkConfig, ContaineredTaskManagerParameters tmParams, String configDirectory, String logDirectory, boolean hasLogback, boolean hasLog4j, boolean hasKrb5, Class<?> mainClass) { final Map<String, String> startCommandValues = new HashMap<>(); startCommandValues.put("java", "$JAVA_HOME/bin/java"); ArrayList<String> params = new ArrayList<>(); params.add(String.format("-Xms%dm", tmParams.taskManagerHeapSizeMB())); params.add(String.format("-Xmx%dm", tmParams.taskManagerHeapSizeMB())); if (tmParams.taskManagerDirectMemoryLimitMB() >= 0) { params.add(String.format("-XX:MaxDirectMemorySize=%dm", tmParams.taskManagerDirectMemoryLimitMB())); } startCommandValues.put("jvmmem", StringUtils.join(params, ' ')); String javaOpts = flinkConfig.getString(CoreOptions.FLINK_JVM_OPTIONS); if (flinkConfig.getString(CoreOptions.FLINK_TM_JVM_OPTIONS).length() > 0) { javaOpts += " " + flinkConfig.getString(CoreOptions.FLINK_TM_JVM_OPTIONS); } if (hasKrb5) { javaOpts += " -Djava.security.krb5.conf=krb5.conf"; } startCommandValues.put("jvmopts", javaOpts); String logging = ""; if (hasLogback || hasLog4j) { logging = "-Dlog.file=" + logDirectory + "/taskmanager.log"; if (hasLogback) { logging += " -Dlogback.configurationFile=file:" + configDirectory + "/logback.xml"; } if (hasLog4j) { logging += " -Dlog4j.configuration=file:" + configDirectory + "/log4j.properties"; } } startCommandValues.put("logging", logging); startCommandValues.put("class", mainClass.getName()); startCommandValues.put("redirects", "1> " + logDirectory + "/taskmanager.out " + "2> " + logDirectory + "/taskmanager.err"); startCommandValues.put("args", "--configDir " + configDirectory); final String commandTemplate = flinkConfig .getString(ConfigConstants.YARN_CONTAINER_START_COMMAND_TEMPLATE, ConfigConstants.DEFAULT_YARN_CONTAINER_START_COMMAND_TEMPLATE); String startCommand = getStartCommand(commandTemplate, startCommandValues); LOG.debug("TaskManager start command: " + startCommand); return startCommand; } private  BootstrapTools(); static ActorSystem startActorSystem(
		Configuration configuration,
		String listeningAddress,
		String portRangeDefinition,
		Logger logger); static ActorSystem startActorSystem(
			Configuration configuration,
			String listeningAddress,
			String portRangeDefinition,
			Logger logger,
			@Nonnull ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static ActorSystem startActorSystem(
			Configuration configuration,
			String actorSystemName,
			String listeningAddress,
			String portRangeDefinition,
			Logger logger,
			@Nonnull ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static ActorSystem startActorSystem(
		Configuration configuration,
		String listeningAddress,
		int listeningPort,
		Logger logger); static ActorSystem startActorSystem(
				Configuration configuration,
				String listeningAddress,
				int listeningPort,
				Logger logger,
				ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static ActorSystem startActorSystem(
		Configuration configuration,
		String actorSystemName,
		String listeningAddress,
		int listeningPort,
		Logger logger,
		ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static Configuration generateTaskManagerConfiguration(
				Configuration baseConfig,
				String jobManagerHostname,
				int jobManagerPort,
				int numSlots,
				FiniteDuration registrationTimeout); static void writeConfiguration(Configuration cfg, File file); static void substituteDeprecatedConfigKey(Configuration config, String deprecated, String designated); static void substituteDeprecatedConfigPrefix(
			Configuration config,
			String deprecatedPrefix,
			String designatedPrefix); static Option newDynamicPropertiesOption(); static Configuration parseDynamicProperties(CommandLine cmd); static String getTaskManagerShellCommand(
			Configuration flinkConfig,
			ContaineredTaskManagerParameters tmParams,
			String configDirectory,
			String logDirectory,
			boolean hasLogback,
			boolean hasLog4j,
			boolean hasKrb5,
			Class<?> mainClass); static String getStartCommand(String template,
		Map<String, String> startCommandValues); static void updateTmpDirectoriesInConfiguration(
			Configuration configuration,
			@Nullable String defaultDirs); static Configuration cloneConfiguration(Configuration configuration); }
BootstrapTools { public static String getTaskManagerShellCommand( Configuration flinkConfig, ContaineredTaskManagerParameters tmParams, String configDirectory, String logDirectory, boolean hasLogback, boolean hasLog4j, boolean hasKrb5, Class<?> mainClass) { final Map<String, String> startCommandValues = new HashMap<>(); startCommandValues.put("java", "$JAVA_HOME/bin/java"); ArrayList<String> params = new ArrayList<>(); params.add(String.format("-Xms%dm", tmParams.taskManagerHeapSizeMB())); params.add(String.format("-Xmx%dm", tmParams.taskManagerHeapSizeMB())); if (tmParams.taskManagerDirectMemoryLimitMB() >= 0) { params.add(String.format("-XX:MaxDirectMemorySize=%dm", tmParams.taskManagerDirectMemoryLimitMB())); } startCommandValues.put("jvmmem", StringUtils.join(params, ' ')); String javaOpts = flinkConfig.getString(CoreOptions.FLINK_JVM_OPTIONS); if (flinkConfig.getString(CoreOptions.FLINK_TM_JVM_OPTIONS).length() > 0) { javaOpts += " " + flinkConfig.getString(CoreOptions.FLINK_TM_JVM_OPTIONS); } if (hasKrb5) { javaOpts += " -Djava.security.krb5.conf=krb5.conf"; } startCommandValues.put("jvmopts", javaOpts); String logging = ""; if (hasLogback || hasLog4j) { logging = "-Dlog.file=" + logDirectory + "/taskmanager.log"; if (hasLogback) { logging += " -Dlogback.configurationFile=file:" + configDirectory + "/logback.xml"; } if (hasLog4j) { logging += " -Dlog4j.configuration=file:" + configDirectory + "/log4j.properties"; } } startCommandValues.put("logging", logging); startCommandValues.put("class", mainClass.getName()); startCommandValues.put("redirects", "1> " + logDirectory + "/taskmanager.out " + "2> " + logDirectory + "/taskmanager.err"); startCommandValues.put("args", "--configDir " + configDirectory); final String commandTemplate = flinkConfig .getString(ConfigConstants.YARN_CONTAINER_START_COMMAND_TEMPLATE, ConfigConstants.DEFAULT_YARN_CONTAINER_START_COMMAND_TEMPLATE); String startCommand = getStartCommand(commandTemplate, startCommandValues); LOG.debug("TaskManager start command: " + startCommand); return startCommand; } private  BootstrapTools(); static ActorSystem startActorSystem(
		Configuration configuration,
		String listeningAddress,
		String portRangeDefinition,
		Logger logger); static ActorSystem startActorSystem(
			Configuration configuration,
			String listeningAddress,
			String portRangeDefinition,
			Logger logger,
			@Nonnull ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static ActorSystem startActorSystem(
			Configuration configuration,
			String actorSystemName,
			String listeningAddress,
			String portRangeDefinition,
			Logger logger,
			@Nonnull ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static ActorSystem startActorSystem(
		Configuration configuration,
		String listeningAddress,
		int listeningPort,
		Logger logger); static ActorSystem startActorSystem(
				Configuration configuration,
				String listeningAddress,
				int listeningPort,
				Logger logger,
				ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static ActorSystem startActorSystem(
		Configuration configuration,
		String actorSystemName,
		String listeningAddress,
		int listeningPort,
		Logger logger,
		ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static Configuration generateTaskManagerConfiguration(
				Configuration baseConfig,
				String jobManagerHostname,
				int jobManagerPort,
				int numSlots,
				FiniteDuration registrationTimeout); static void writeConfiguration(Configuration cfg, File file); static void substituteDeprecatedConfigKey(Configuration config, String deprecated, String designated); static void substituteDeprecatedConfigPrefix(
			Configuration config,
			String deprecatedPrefix,
			String designatedPrefix); static Option newDynamicPropertiesOption(); static Configuration parseDynamicProperties(CommandLine cmd); static String getTaskManagerShellCommand(
			Configuration flinkConfig,
			ContaineredTaskManagerParameters tmParams,
			String configDirectory,
			String logDirectory,
			boolean hasLogback,
			boolean hasLog4j,
			boolean hasKrb5,
			Class<?> mainClass); static String getStartCommand(String template,
		Map<String, String> startCommandValues); static void updateTmpDirectoriesInConfiguration(
			Configuration configuration,
			@Nullable String defaultDirs); static Configuration cloneConfiguration(Configuration configuration);  }