boolean exec(TaskSpec spec, PrintStream out) throws Exception {
        TaskController controller;
        try {
            controller = spec.newController(EXEC_TASK_ID);
        } catch (Exception e) {
            out.println("Unable to create the task controller.");
            e.printStackTrace(out);
            return false;
        }
        Set<String> nodes = controller.targetNodes(platform.topology());
        if (!nodes.contains(platform.curNode().name())) {
            out.println("This task is not configured to run on this node.  It runs on node(s): " +
                String.join(", ", nodes) + ", whereas this node is " +
                platform.curNode().name());
            return false;
        }
        KafkaFuture<String> future;
        try {
            future = workerManager.createWorker(EXEC_WORKER_ID, EXEC_TASK_ID, spec);
        } catch (Throwable e) {
            out.println("createWorker failed");
            e.printStackTrace(out);
            return false;
        }
        out.println("Waiting for completion of task:" + JsonUtil.toPrettyJsonString(spec));
        String error = future.get();
        if (error == null || error.isEmpty()) {
            out.println("Task succeeded with status " +
                JsonUtil.toPrettyJsonString(workerManager.workerStates().get(EXEC_WORKER_ID).status()));
            return true;
        } else {
            out.println("Task failed with status " +
                JsonUtil.toPrettyJsonString(workerManager.workerStates().get(EXEC_WORKER_ID).status()) +
                " and error " + error);
            return false;
        }
    }
Start a task on the agent, and block until it completes.

@param spec          The task specifiction.
@param out           The output stream to print to.

@return              True if the task run successfully; false otherwise.
trogdor/src/main/java/org/apache/kafka/trogdor/agent/Agent.java