private void handleWorkerCompletion(ManagedTask task, String nodeName, WorkerDone state) {
        if (state.error().isEmpty()) {
            log.info("{}: Worker {} finished with status '{}'",
                nodeName, task.id, JsonUtil.toJsonString(state.status()));
        } else {
            log.warn("{}: Worker {} finished with error '{}' and status '{}'",
                nodeName, task.id, state.error(), JsonUtil.toJsonString(state.status()));
            task.maybeSetError(state.error());
        }
        TreeMap<String, Long> activeWorkerIds = task.activeWorkerIds();
        if (activeWorkerIds.isEmpty()) {
            task.doneMs = time.milliseconds();
            task.state = TaskStateType.DONE;
            log.info("{}: Task {} is now complete on {} with error: {}",
                nodeName, task.id, String.join(", ", task.workerIds.keySet()),
                task.error.isEmpty() ? "(none)" : task.error);
        } else if ((task.state == TaskStateType.RUNNING) && (!task.error.isEmpty())) {
            log.info("{}: task {} stopped with error {}.  Stopping worker(s): {}",
                nodeName, task.id, task.error, Utils.mkString(activeWorkerIds, "{", "}", ": ", ", "));
            task.state = TaskStateType.STOPPING;
            for (Map.Entry<String, Long> entry : activeWorkerIds.entrySet()) {
                nodeManagers.get(entry.getKey()).stopWorker(entry.getValue());
            }
        }
    }
Handle a worker being completed.

@param task      The task that owns the worker.
@param nodeName  The name of the node on which the worker is running.
@param state     The worker state.
trogdor/src/main/java/org/apache/kafka/trogdor/coordinator/TaskManager.java