@Override
        public Void call() throws Exception {
            if (id.isEmpty()) {
                throw new InvalidRequestException("Invalid empty ID in stopTask request.");
            }
            ManagedTask task = tasks.get(id);
            if (task == null) {
                log.info("Can't cancel non-existent task {}.", id);
                return null;
            }
            switch (task.state) {
                case PENDING:
                    task.cancelled = true;
                    task.clearStartFuture();
                    task.doneMs = time.milliseconds();
                    task.state = TaskStateType.DONE;
                    log.info("Stopped pending task {}.", id);
                    break;
                case RUNNING:
                    task.cancelled = true;
                    TreeMap<String, Long> activeWorkerIds = task.activeWorkerIds();
                    if (activeWorkerIds.isEmpty()) {
                        if (task.error.isEmpty()) {
                            log.info("Task {} is now complete with no errors.", id);
                        } else {
                            log.info("Task {} is now complete with error: {}", id, task.error);
                        }
                        task.doneMs = time.milliseconds();
                        task.state = TaskStateType.DONE;
                    } else {
                        for (Map.Entry<String, Long> entry : activeWorkerIds.entrySet()) {
                            nodeManagers.get(entry.getKey()).stopWorker(entry.getValue());
                        }
                        log.info("Cancelling task {} with worker(s) {}",
                            id, Utils.mkString(activeWorkerIds, "", "", " = ", ", "));
                        task.state = TaskStateType.STOPPING;
                    }
                    break;
                case STOPPING:
                    log.info("Can't cancel task {} because it is already stopping.", id);
                    break;
                case DONE:
                    log.info("Can't cancel task {} because it is already done.", id);
                    break;
            }
            return null;
        }
Handles cancelling a task.  Processed by the state change thread.
trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/TaskManager.java