public KafkaFuture<String> createWorker(long workerId, String taskId, TaskSpec spec) throws Throwable {
        try (ShutdownManager.Reference ref = shutdownManager.takeReference()) {
            final Worker worker = stateChangeExecutor.
                submit(new CreateWorker(workerId, taskId, spec, time.milliseconds())).get();
            if (worker.doneFuture != null) {
                log.info("{}: Ignoring request to create worker {}, because there is already " +
                    "a worker with that id.", nodeName, workerId);
                return worker.doneFuture;
            }
            worker.doneFuture = new KafkaFutureImpl<>();
            if (worker.spec.endMs() <= time.milliseconds()) {
                log.info("{}: Will not run worker {} as it has expired.", nodeName, worker);
                stateChangeExecutor.submit(new HandleWorkerHalting(worker,
                    "worker expired", true));
                return worker.doneFuture;
            }
            KafkaFutureImpl<String> haltFuture = new KafkaFutureImpl<>();
            haltFuture.thenApply((KafkaFuture.BaseFunction<String, Void>) errorString -> {
                if (errorString == null)
                    errorString = "";
                if (errorString.isEmpty()) {
                    log.info("{}: Worker {} is halting.", nodeName, worker);
                } else {
                    log.info("{}: Worker {} is halting with error {}",
                        nodeName, worker, errorString);
                }
                stateChangeExecutor.submit(
                    new HandleWorkerHalting(worker, errorString, false));
                return null;
            });
            try {
                worker.taskWorker.start(platform, worker.status, haltFuture);
            } catch (Exception e) {
                log.info("{}: Worker {} start() exception", nodeName, worker, e);
                stateChangeExecutor.submit(new HandleWorkerHalting(worker,
                    "worker.start() exception: " + Utils.stackTrace(e), true));
            }
            stateChangeExecutor.submit(new FinishCreatingWorker(worker));
            return worker.doneFuture;
        } catch (ExecutionException e) {
            if (e.getCause() instanceof RequestConflictException) {
                log.info("{}: request conflict while creating worker {} for task {} with spec {}.",
                    nodeName, workerId, taskId, spec);
            } else {
                log.info("{}: Error creating worker {} for task {} with spec {}",
                    nodeName, workerId, taskId, spec, e);
            }
            throw e.getCause();
        }
    }
Whether we should destroy the records of this worker once it stops.
trogdor/src/main/java/org/apache/kafka/trogdor/agent/WorkerManager.java