@Override
    public void kill() throws IOException {
        LOG.info("Killing {}:{}", supervisorId, workerId);
        if (shutdownTimer == null) {
            shutdownTimer = shutdownDuration.time();
        }
        try {
            if (resourceIsolationManager != null) {
                resourceIsolationManager.kill(getWorkerUser(), workerId);
            }
        } catch (IOException e) {
            numKillExceptions.mark();
            throw e;
        }
    }
@Test
    public void testKill() throws Exception {
        final String topoId = "test_topology";
        final Map<String, Object> superConf = new HashMap<>();
        AdvancedFSOps ops = mock(AdvancedFSOps.class);
        when(ops.doRequiredTopoFilesExist(superConf, topoId)).thenReturn(true);
        LocalAssignment la = new LocalAssignment();
        la.set_topology_id(topoId);
        MockResourceIsolationManager iso = new MockResourceIsolationManager();
        String workerId = "worker-id";
        MockContainer mc = new MockContainer(ContainerType.LAUNCH, superConf,
            "SUPERVISOR", 6628, 8080, la, iso, workerId, new HashMap<>(), ops, new StormMetricsRegistry());
        iso.allWorkerIds.add(workerId);

        assertEquals(Collections.EMPTY_LIST, iso.killedWorkerIds);
        assertEquals(Collections.EMPTY_LIST, iso.forceKilledWorkerIds);

        mc.kill();
        assertEquals(iso.allWorkerIds, iso.killedWorkerIds);
        assertEquals(Collections.EMPTY_LIST, iso.forceKilledWorkerIds);
        iso.killedWorkerIds.clear();

        mc.forceKill();
        assertEquals(Collections.EMPTY_LIST, iso.killedWorkerIds);
        assertEquals(iso.allWorkerIds, iso.forceKilledWorkerIds);
    }