@Override
        public Void call() throws Exception {
            if (id.isEmpty()) {
                throw new InvalidRequestException("Invalid empty ID in createTask request.");
            }
            ManagedTask task = tasks.get(id);
            if (task != null) {
                if (!task.originalSpec.equals(originalSpec)) {
                    throw new RequestConflictException("Task ID " + id + " already " +
                        "exists, and has a different spec " + task.originalSpec);
                }
                log.info("Task {} already exists with spec {}", id, originalSpec);
                return null;
            }
            TaskController controller = null;
            String failure = null;
            try {
                controller = spec.newController(id);
            } catch (Throwable t) {
                failure = "Failed to create TaskController: " + t.getMessage();
            }
            if (failure != null) {
                log.info("Failed to create a new task {} with spec {}: {}",
                    id, spec, failure);
                task = new ManagedTask(id, originalSpec, spec, null, TaskStateType.DONE);
                task.doneMs = time.milliseconds();
                task.maybeSetError(failure);
                tasks.put(id, task);
                return null;
            }
            task = new ManagedTask(id, originalSpec, spec, controller, TaskStateType.PENDING);
            tasks.put(id, task);
            long delayMs = task.startDelayMs(time.milliseconds());
            task.startFuture = scheduler.schedule(executor, new RunTask(task), delayMs);
            log.info("Created a new task {} with spec {}, scheduled to start {} ms from now.",
                    id, spec, delayMs);
            return null;
        }
Handles a request to create a new task.  Processed by the state change thread.
trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/TaskManager.java