public void launchWorker ( Map conf , IContext sharedContext , String topologyId , String supervisorId , Integer port , String workerId , LocalAssignment assignment ) throws IOException { // get supervisor conf Map stormConf = StormConfig . read_supervisor_topology_conf ( conf , topologyId ) ; String stormHome = System . getProperty ( "jstorm.home" ) ; if ( StringUtils . isBlank ( stormHome ) ) { stormHome = "./" ; } // get worker conf String topologyRoot = StormConfig . supervisor_stormdist_root ( conf , topologyId ) ; Map workerConf = StormConfig . read_topology_conf ( topologyRoot , topologyId ) ; Map totalConf = new HashMap ( ) ; totalConf . putAll ( conf ) ; totalConf . putAll ( stormConf ) ; totalConf . putAll ( workerConf ) ; /** * don't use environment.putAll(System.getenv); */ Map < String , String > environment = new HashMap <> ( ) ; if ( ConfigExtension . getWorkerRedirectOutput ( totalConf ) ) { environment . put ( "REDIRECT" , "true" ) ; } else { environment . put ( "REDIRECT" , "false" ) ; } environment . put ( "LD_LIBRARY_PATH" , ( String ) totalConf . get ( Config . JAVA_LIBRARY_PATH ) ) ; environment . put ( "jstorm.home" , stormHome ) ; environment . put ( "jstorm.workerId" , workerId ) ; environment . put ( "jstorm.on.yarn" , isJstormOnYarn ? "1" : "0" ) ; String launcherCmd = getLauncherParameter ( assignment , totalConf , stormHome , topologyId , port ) ; String workerCmd = getWorkerParameter ( assignment , totalConf , stormHome , topologyId , supervisorId , workerId , port ) ; String cmd = launcherCmd + " " + workerCmd ; cmd = cmd . replace ( "%JSTORM_HOME%" , stormHome ) ; LOG . info ( "Launching worker with command: " + cmd ) ; LOG . info ( "Environment:" + environment . toString ( ) ) ; /** * if run on yarn, set backend false, otherwise set true */ boolean backend = ! isJstormOnYarn ; LOG . info ( "backend mode is " + backend ) ; JStormUtils . launchProcess ( cmd , environment , backend ) ; }
launch a worker in distributed mode