@Override
        public Worker call() throws Exception {
            try {
                Worker worker = workers.get(workerId);
                if (worker != null) {
                    if (!worker.taskId().equals(taskId)) {
                        throw new RequestConflictException("There is already a worker ID " + workerId +
                            " with a different task ID.");
                    } else if (!worker.spec().equals(spec)) {
                        throw new RequestConflictException("There is already a worker ID " + workerId +
                            " with a different task spec.");
                    } else {
                        return worker;
                    }
                }
                worker = new Worker(workerId, taskId, spec, now);
                workers.put(workerId, worker);
                log.info("{}: Created worker {} with spec {}", nodeName, worker, spec);
                return worker;
            } catch (Exception e) {
                log.info("{}: unable to create worker {} for task {}, with spec {}",
                    nodeName, workerId, taskId, spec, e);
                throw e;
            }
        }
Handles a request to create a new worker.  Processed by the state change thread.
trogdor/src/main/java/org/apache/kafka/trogdor/agent/WorkerManager.java