@Override
        public Void call() throws Exception {
            if (id.isEmpty()) {
                throw new InvalidRequestException("Invalid empty ID in destroyTask request.");
            }
            ManagedTask task = tasks.remove(id);
            if (task == null) {
                log.info("Can't destroy task {}: no such task found.", id);
                return null;
            }
            log.info("Destroying task {}.", id);
            task.clearStartFuture();
            for (Map.Entry<String, Long> entry : task.workerIds.entrySet()) {
                long workerId = entry.getValue();
                workerStates.remove(workerId);
                String nodeName = entry.getKey();
                nodeManagers.get(nodeName).destroyWorker(workerId);
            }
            return null;
        }
Handles destroying a task.  Processed by the state change thread.
trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/TaskManager.java