@Override
    public void start(Platform platform, WorkerStatusTracker status,
                      KafkaFutureImpl<String> doneFuture) throws Exception {
        if (!running.compareAndSet(false, true)) {
            throw new IllegalStateException("ConsumeBenchWorker is already running.");
        }
        log.info("{}: Activating ExternalCommandWorker with {}", id, spec);
        this.status = status;
        this.doneFuture = doneFuture;
        this.executor = Executors.newCachedThreadPool(
            ThreadUtils.createThreadFactory("ExternalCommandWorkerThread%d", false));
        Process process;
        try {
            process = startProcess();
        } catch (Throwable t) {
            log.error("{}: Unable to start process", id, t);
            executor.shutdown();
            doneFuture.complete("Unable to start process: " + t.getMessage());
            return;
        }
        Future<?> stdoutFuture = executor.submit(new StdoutMonitor(process));
        Future<?> stderrFuture = executor.submit(new StderrMonitor(process));
        executor.submit(new StdinWriter(process));
        Future<?> terminatorFuture = executor.submit(new Terminator(process));
        executor.submit(new ExitMonitor(process, stdoutFuture, stderrFuture, terminatorFuture));
        ObjectNode startMessage = new ObjectNode(JsonNodeFactory.instance);
        startMessage.set("id", new TextNode(id));
        startMessage.set("workload", spec.workload());
        stdinQueue.add(Optional.of(startMessage));
    }
The executor service for this worker.
trogdor/src/main/java/org/apache/kafka/trogdor/workload/ExternalCommandWorker.java