@Override
        public Void call() throws Exception {
            Worker worker = workers.get(workerId);
            if (worker == null) {
                log.info("{}: Can't stop worker {} because there is no worker with that ID.",
                    nodeName, workerId);
                return null;
            }
            if (mustDestroy) {
                worker.mustDestroy = true;
            }
            switch (worker.state) {
                case STARTING:
                    log.info("{}: Cancelling worker {} during its startup process.",
                        nodeName, worker);
                    worker.state = State.CANCELLING;
                    break;
                case CANCELLING:
                    log.info("{}: Can't stop worker {}, because it is already being " +
                        "cancelled.", nodeName, worker);
                    break;
                case RUNNING:
                    log.info("{}: Stopping running worker {}.", nodeName, worker);
                    worker.transitionToStopping();
                    break;
                case STOPPING:
                    log.info("{}: Can't stop worker {}, because it is already " +
                            "stopping.", nodeName, worker);
                    break;
                case DONE:
                    if (worker.mustDestroy) {
                        log.info("{}: destroying worker {} with error {}",
                            nodeName, worker, worker.error);
                        workers.remove(worker.workerId);
                    } else {
                        log.debug("{}: Can't stop worker {}, because it is already done.",
                            nodeName, worker);
                    }
                    break;
            }
            return null;
        }
Stops a worker.  Processed by the state change thread.
trogdor/src/main/java/org/apache/kafka/trogdor/agent/WorkerManager.java