@Override
    public void launch() throws IOException {
        type.assertFull();
        String numaId = SupervisorUtils.getNumaIdForPort(port, conf);
        if (numaId == null) {
            LOG.info("Launching worker with assignment {} for this supervisor {} on port {} with id {}",
                    assignment, supervisorId, port, workerId);
        } else {
            LOG.info("Launching worker with assignment {} for this supervisor {} on port {} with id {}  bound to numa zone {}",
                    assignment, supervisorId, port, workerId, numaId);
        }
        exitedEarly = false;

        final WorkerResources resources = assignment.get_resources();
        final int memOnHeap = getMemOnHeap(resources);
        final int memOffHeap = getMemOffHeap(resources);
        memoryLimitMb = calculateMemoryLimit(resources, memOnHeap);
        final String stormRoot = ConfigUtils.supervisorStormDistRoot(conf, topologyId);
        String jlp = javaLibraryPath(stormRoot, conf);

        Map<String, String> topEnvironment = new HashMap<String, String>();
        @SuppressWarnings("unchecked")
        Map<String, String> environment = (Map<String, String>) topoConf.get(Config.TOPOLOGY_ENVIRONMENT);
        if (environment != null) {
            topEnvironment.putAll(environment);
        }

        String ldLibraryPath = topEnvironment.get("LD_LIBRARY_PATH");
        if (ldLibraryPath != null) {
            jlp = jlp + System.getProperty("path.separator") + ldLibraryPath;
        }

        topEnvironment.put("LD_LIBRARY_PATH", jlp);

        if (resourceIsolationManager.isResourceManaged()) {
            final int cpu = (int) Math.ceil(resources.get_cpu());
            //Save the memory limit so we can enforce it less strictly
            resourceIsolationManager.reserveResourcesForWorker(workerId, (int) memoryLimitMb, cpu, numaId);
        }

        List<String> commandList = mkLaunchCommand(memOnHeap, memOffHeap, stormRoot, jlp, numaId);

        LOG.info("Launching worker with command: {}. ", ServerUtils.shellCmd(commandList));

        String workerDir = ConfigUtils.workerRoot(conf, workerId);

        String logPrefix = "Worker Process " + workerId;
        ProcessExitCallback processExitCallback = new ProcessExitCallback(logPrefix);
        resourceIsolationManager.launchWorkerProcess(getWorkerUser(), topologyId, topoConf, port, workerId,
            commandList, topEnvironment, logPrefix, processExitCallback, new File(workerDir));
    }
@Test
    public void testLaunchStorm0version() throws Exception {
        final String topoId = "test_topology_storm_0.x";
        final int supervisorPort = 6628;
        final int port = 8080;
        final String stormHome = ContainerTest.asAbsPath("tmp", "storm-home");
        final String stormLogDir = ContainerTest.asFile(".", "target").getCanonicalPath();
        final String workerId = "worker-id";
        final String stormLocal = ContainerTest.asAbsPath("tmp", "storm-local");
        final String distRoot = ContainerTest.asAbsPath(stormLocal, "supervisor", "stormdist", topoId);
        final File stormcode = new File(distRoot, "stormcode.ser");
        final File stormjar = new File(distRoot, "stormjar.jar");
        final String log4jdir = ContainerTest.asAbsPath(stormHome, "conf");
        final String workerConf = ContainerTest.asAbsPath(log4jdir, "worker.xml");
        final String workerRoot = ContainerTest.asAbsPath(stormLocal, "workers", workerId);
        final String workerTmpDir = ContainerTest.asAbsPath(workerRoot, "tmp");

        final StormTopology st = new StormTopology();
        st.set_spouts(new HashMap<>());
        st.set_bolts(new HashMap<>());
        st.set_state_spouts(new HashMap<>());

        // minimum 0.x version of supporting STORM-2448 would be 0.10.3
        st.set_storm_version("0.10.3");
        byte[] serializedState = Utils.gzip(Utils.thriftSerialize(st));

        final Map<String, Object> superConf = new HashMap<>();
        superConf.put(Config.STORM_LOCAL_DIR, stormLocal);
        superConf.put(Config.STORM_WORKERS_ARTIFACTS_DIR, stormLocal);
        superConf.put(DaemonConfig.STORM_LOG4J2_CONF_DIR, log4jdir);
        superConf.put(Config.WORKER_CHILDOPTS, " -Dtesting=true");

        LocalAssignment la = new LocalAssignment();
        la.set_topology_id(topoId);

        AdvancedFSOps ops = mock(AdvancedFSOps.class);
        when(ops.doRequiredTopoFilesExist(superConf, topoId)).thenReturn(true);
        when(ops.slurp(stormcode)).thenReturn(serializedState);

        LocalState ls = mock(LocalState.class);

        MockResourceIsolationManager iso = new MockResourceIsolationManager();

        checkpoint(() -> {
                       MockBasicContainer mc = new MockBasicContainer(ContainerType.LAUNCH, superConf,
                "SUPERVISOR", supervisorPort, port, la, iso, ls, workerId, new StormMetricsRegistry(),
                new HashMap<>(), ops, "profile");

                       mc.launch();

                       assertEquals(1, iso.workerCmds.size());
                       CommandRun cmd = iso.workerCmds.get(0);
                       iso.workerCmds.clear();
                       assertListEquals(Arrays.asList(
                           "java",
                           "-cp",
                           "FRAMEWORK_CP:" + stormjar.getAbsolutePath(),
                           "-Dlogging.sensitivity=S3",
                           "-Dlogfile.name=worker.log",
                           "-Dstorm.home=" + stormHome,
                           "-Dworkers.artifacts=" + stormLocal,
                           "-Dstorm.id=" + topoId,
                           "-Dworker.id=" + workerId,
                           "-Dworker.port=" + port,
                           "-Dstorm.log.dir=" + stormLogDir,
                           "-DLog4jContextSelector=org.apache.logging.log4j.core.selector.BasicContextSelector",
                           "-Dstorm.local.dir=" + stormLocal,
                           "-Dworker.memory_limit_mb=768",
                           "-Dlog4j.configurationFile=" + workerConf,
                           "backtype.storm.LogWriter",
                           "java",
                           "-server",
                           "-Dlogging.sensitivity=S3",
                           "-Dlogfile.name=worker.log",
                           "-Dstorm.home=" + stormHome,
                           "-Dworkers.artifacts=" + stormLocal,
                           "-Dstorm.id=" + topoId,
                           "-Dworker.id=" + workerId,
                           "-Dworker.port=" + port,
                           "-Dstorm.log.dir=" + stormLogDir,
                           "-DLog4jContextSelector=org.apache.logging.log4j.core.selector.BasicContextSelector",
                           "-Dstorm.local.dir=" + stormLocal,
                           "-Dworker.memory_limit_mb=768",
                           "-Dlog4j.configurationFile=" + workerConf,
                           "-Dtesting=true",
                           "-Djava.library.path=JLP",
                           "-Dstorm.conf.file=",
                           "-Dstorm.options=",
                           "-Djava.io.tmpdir=" + workerTmpDir,
                           "-cp",
                           "FRAMEWORK_CP:" + stormjar.getAbsolutePath(),
                           "backtype.storm.daemon.worker",
                           topoId,
                           "SUPERVISOR",
                           String.valueOf(port),
                           workerId
                       ), cmd.cmd);
                       assertEquals(new File(workerRoot), cmd.pwd);
                   },
                   ConfigUtils.STORM_HOME, stormHome,
                   "storm.log.dir", stormLogDir);
    }