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 ) ; } //applicable only for YarnMiniCluster secure test run //krb5.conf file will be available as local resource in JM/TM container 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 ; }
Generates the shell command to start a task manager .