@Test
    public void testTaskRequest() throws Exception {
        MockTime time = new MockTime(0, 0, 0);
        Scheduler scheduler = new MockScheduler(time);
        try (MiniTrogdorCluster cluster = new MiniTrogdorCluster.Builder().
            addCoordinator("node01").
            addAgent("node02").
            scheduler(scheduler).
            build()) {
            CoordinatorClient coordinatorClient = cluster.coordinatorClient();

            NoOpTaskSpec fooSpec = new NoOpTaskSpec(1, 10);
            coordinatorClient.createTask(new CreateTaskRequest("foo", fooSpec));
            TaskState expectedState = new ExpectedTaskBuilder("foo").taskState(new TaskPending(fooSpec)).build().taskState();

            TaskState resp = coordinatorClient.task(new TaskRequest("foo"));
            assertEquals(expectedState, resp);

            time.sleep(2);
            new ExpectedTasks().
                addTask(new ExpectedTaskBuilder("foo").
                    taskState(new TaskRunning(fooSpec, 2, new TextNode("active"))).
                    workerState(new WorkerRunning("foo", fooSpec, 2, new TextNode("active"))).
                    build()).
                waitFor(coordinatorClient).
                waitFor(cluster.agentClient("node02"));

            assertThrows(NotFoundException.class, () -> coordinatorClient.task(new TaskRequest("non-existent-foo")));
        }
    }
If an agent fails in the middle of a task and comes back up when the task is considered expired,
we want the task to be marked as DONE and not re-sent should a second failure happen.
trogdor/src/test/java/org/apache/kafka/trogdor/coordinator/CoordinatorTest.java