@Override
    public boolean runProfiling(ProfileRequest request, boolean stop) throws IOException, InterruptedException {
        type.assertFull();
        String targetDir = ConfigUtils.workerArtifactsRoot(conf, topologyId, port);

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

        String str = ConfigUtils.workerArtifactsPidPath(conf, topologyId, port);

        String workerPid = ops.slurpString(new File(str)).trim();

        ProfileAction profileAction = request.get_action();
        String logPrefix = "ProfilerAction process " + topologyId + ":" + port + " PROFILER_ACTION: " + profileAction
                           + " ";

        List<String> command = mkProfileCommand(profileAction, stop, workerPid, targetDir);

        File targetFile = new File(targetDir);
        if (command.size() > 0) {
            return resourceIsolationManager.runProfilingCommand(getWorkerUser(), workerId, command, env, logPrefix, targetFile);
        }
        LOG.warn("PROFILING REQUEST NOT SUPPORTED {} IGNORED...", request);
        return true;
    }
@Test
    public void testRunProfiling() throws Exception {
        final long pid = 100;
        final String topoId = "test_topology";
        final int supervisorPort = 6628;
        final int port = 8080;
        final String workerId = "worker-id";
        final String stormLocal = ContainerTest.asAbsPath("tmp", "testing");
        final String topoRoot = ContainerTest.asAbsPath(stormLocal, topoId, String.valueOf(port));
        final File workerArtifactsPid = ContainerTest.asAbsFile(topoRoot, "worker.pid");

        final Map<String, Object> superConf = new HashMap<>();
        superConf.put(Config.STORM_LOCAL_DIR, stormLocal);
        superConf.put(Config.STORM_WORKERS_ARTIFACTS_DIR, stormLocal);

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

        AdvancedFSOps ops = mock(AdvancedFSOps.class);
        when(ops.doRequiredTopoFilesExist(superConf, topoId)).thenReturn(true);
        when(ops.slurpString(workerArtifactsPid)).thenReturn(String.valueOf(pid));

        LocalState ls = mock(LocalState.class);

        MockResourceIsolationManager iso = new MockResourceIsolationManager();

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

        //HEAP DUMP
        ProfileRequest req = new ProfileRequest();
        req.set_action(ProfileAction.JMAP_DUMP);

        mc.runProfiling(req, false);

        assertEquals(1, iso.profileCmds.size());
        CommandRun cmd = iso.profileCmds.get(0);
        iso.profileCmds.clear();
        assertEquals(Arrays.asList("profile", String.valueOf(pid), "jmap", topoRoot), cmd.cmd);
        assertEquals(new File(topoRoot), cmd.pwd);

        // JSTACK DUMP
        req.set_action(ProfileAction.JSTACK_DUMP);

        mc.runProfiling(req, false);

        assertEquals(1, iso.profileCmds.size());
        cmd = iso.profileCmds.get(0);
        iso.profileCmds.clear();
        assertEquals(Arrays.asList("profile", String.valueOf(pid), "jstack", topoRoot), cmd.cmd);
        assertEquals(new File(topoRoot), cmd.pwd);

        //RESTART
        req.set_action(ProfileAction.JVM_RESTART);

        mc.runProfiling(req, false);

        assertEquals(1, iso.profileCmds.size());
        cmd = iso.profileCmds.get(0);
        iso.profileCmds.clear();
        assertEquals(Arrays.asList("profile", String.valueOf(pid), "kill"), cmd.cmd);
        assertEquals(new File(topoRoot), cmd.pwd);

        //JPROFILE DUMP
        req.set_action(ProfileAction.JPROFILE_DUMP);

        mc.runProfiling(req, false);

        assertEquals(1, iso.profileCmds.size());
        cmd = iso.profileCmds.get(0);
        iso.profileCmds.clear();
        assertEquals(Arrays.asList("profile", String.valueOf(pid), "dump", topoRoot), cmd.cmd);
        assertEquals(new File(topoRoot), cmd.pwd);

        //JPROFILE START
        req.set_action(ProfileAction.JPROFILE_STOP);

        mc.runProfiling(req, false);

        assertEquals(1, iso.profileCmds.size());
        cmd = iso.profileCmds.get(0);
        iso.profileCmds.clear();
        assertEquals(Arrays.asList("profile", String.valueOf(pid), "start"), cmd.cmd);
        assertEquals(new File(topoRoot), cmd.pwd);

        //JPROFILE STOP
        req.set_action(ProfileAction.JPROFILE_STOP);

        mc.runProfiling(req, true);

        assertEquals(1, iso.profileCmds.size());
        cmd = iso.profileCmds.get(0);
        iso.profileCmds.clear();
        assertEquals(Arrays.asList("profile", String.valueOf(pid), "stop", topoRoot), cmd.cmd);
        assertEquals(new File(topoRoot), cmd.pwd);
    }