@Override
        public Void call() throws Exception {
            log.info("{}: Destroying all workers.", nodeName);

            // StopWorker may remove elements from the set of worker IDs.  That might generate
            // a ConcurrentModificationException if we were iterating over the worker ID
            // set directly.  Therefore, we make a copy of the worker IDs here and iterate
            // over that instead.
            //
            // Note that there is no possible way that more worker IDs can be added while this
            // callable is running, because the state change executor is single-threaded.
            ArrayList<Long> workerIds = new ArrayList<>(workers.keySet());

            for (long workerId : workerIds) {
                try {
                    new StopWorker(workerId, true).call();
                } catch (Exception e) {
                    log.error("Failed to stop worker {}", workerId, e);
                }
            }
            return null;
        }
Begins the process of destroying all workers.  Processed by the state change thread.
trogdor/src/main/java/org/apache/kafka/trogdor/agent/WorkerManager.java