@Override
        public Void call() throws Exception {
            try {
                WorkerState prevState = workerStates.get(workerId);
                if (prevState == null) {
                    throw new RuntimeException("Unable to find workerId " + workerId);
                }
                ManagedTask task = tasks.get(prevState.taskId());
                if (task == null) {
                    throw new RuntimeException("Unable to find taskId " + prevState.taskId());
                }
                log.debug("Task {}: Updating worker state for {} on {} from {} to {}.",
                    task.id, workerId, nodeName, prevState, nextState);
                workerStates.put(workerId, nextState);
                if (nextState.done() && (!prevState.done())) {
                    handleWorkerCompletion(task, nodeName, (WorkerDone) nextState);
                }
            } catch (Exception e) {
                log.error("Error updating worker state for {} on {}.  Stopping worker.",
                    workerId, nodeName, e);
                nodeManagers.get(nodeName).stopWorker(workerId);
            }
            return null;
        }
Updates the state of a worker.  Process by the state change thread.
trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/TaskManager.java