@Override
        public Void call() throws Exception {
            task.clearStartFuture();
            if (task.state != TaskStateType.PENDING) {
                log.info("Can't start task {}, because it is already in state {}.",
                    task.id, task.state);
                return null;
            }
            TreeSet<String> nodeNames;
            try {
                nodeNames = task.findNodeNames();
            } catch (Exception e) {
                log.error("Unable to find nodes for task {}", task.id, e);
                task.doneMs = time.milliseconds();
                task.state = TaskStateType.DONE;
                task.maybeSetError("Unable to find nodes for task: " + e.getMessage());
                return null;
            }
            log.info("Running task {} on node(s): {}", task.id, String.join(", ", nodeNames));
            task.state = TaskStateType.RUNNING;
            task.startedMs = time.milliseconds();
            for (String workerName : nodeNames) {
                long workerId = nextWorkerId++;
                task.workerIds.put(workerName, workerId);
                workerStates.put(workerId, new WorkerReceiving(task.id, task.spec));
                nodeManagers.get(workerName).createWorker(workerId, task.id, task.spec);
            }
            return null;
        }
Handles starting a task.  Processed by the state change thread.
trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/TaskManager.java