@Override
    public void cleanUp() throws IOException {
        try (Timer.Context t = cleanupDuration.time()) {
            containerMemoryTracker.remove(port);
            cleanUpForRestart();
        } catch (IOException e) {
            //This may or may not be reported depending on when process exits
            numCleanupExceptions.mark();
            throw e;
        }
    }
@Test
    public void testCleanup() throws Exception {
        final int supervisorPort = 6628;
        final int port = 8080;
        final String topoId = "test_topology";
        final String workerId = "worker_id";
        final String user = "me";
        final String stormLocal = asAbsPath("tmp", "testing");
        final File workerArtifacts = asAbsFile(stormLocal, topoId, String.valueOf(port));
        final File logMetadataFile = new File(workerArtifacts, "worker.yaml");
        final File workerUserFile = asAbsFile(stormLocal, "workers-users", workerId);
        final File workerRoot = asAbsFile(stormLocal, "workers", workerId);

        final Map<String, Object> topoConf = new HashMap<>();

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

        final StringWriter yamlDump = new StringWriter();

        AdvancedFSOps ops = mock(AdvancedFSOps.class);
        when(ops.doRequiredTopoFilesExist(superConf, topoId)).thenReturn(true);
        when(ops.fileExists(workerArtifacts)).thenReturn(true);
        when(ops.fileExists(workerRoot)).thenReturn(true);
        when(ops.getWriter(logMetadataFile)).thenReturn(yamlDump);

        ResourceIsolationInterface iso = mock(ResourceIsolationInterface.class);
        when(iso.isResourceManaged()).thenReturn(true);

        LocalAssignment la = new LocalAssignment();
        la.set_owner(user);
        la.set_topology_id(topoId);
        MockContainer mc = new MockContainer(ContainerType.LAUNCH, superConf,
                                             "SUPERVISOR", supervisorPort, port, la, iso, workerId, topoConf, ops, new StormMetricsRegistry());

        mc.cleanUp();
        verify(iso).cleanup(user, workerId, port);

        verify(ops).deleteIfExists(eq(new File(workerRoot, "pids")), eq(user), anyString());
        verify(ops).deleteIfExists(eq(new File(workerRoot, "tmp")), eq(user), anyString());
        verify(ops).deleteIfExists(eq(new File(workerRoot, "heartbeats")), eq(user), anyString());
        verify(ops).deleteIfExists(eq(workerRoot), eq(user), anyString());
        verify(ops).deleteIfExists(workerUserFile);
    }