@Test
    public void testAgentFailureAndTaskExpiry() 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, 500);
            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"));

            cluster.restartAgent("node02");
            time.sleep(550);
            // coordinator heartbeat sees that the agent is back up, re-schedules the task but the agent expires it
            new ExpectedTasks().
                addTask(new ExpectedTaskBuilder("foo").
                    taskState(new TaskDone(fooSpec, 2, 552, "worker expired", false, null)).
                    workerState(new WorkerDone("foo", fooSpec, 552, 552, null, "worker expired")).
                    build()).
                waitFor(coordinatorClient).
                waitFor(cluster.agentClient("node02"));

            cluster.restartAgent("node02");
            // coordinator heartbeat sees that the agent is back up but does not re-schedule the task as it is DONE
            new ExpectedTasks().
                addTask(new ExpectedTaskBuilder("foo").
                    taskState(new TaskDone(fooSpec, 2, 552, "worker expired", false, null)).
                    // no worker states
                    build()).
                waitFor(coordinatorClient).
                waitFor(cluster.agentClient("node02"));
        }
    }
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