@Test
    public void testAgentCreateWorkers() throws Exception {
        MockTime time = new MockTime(0, 0, 0);
        MockScheduler scheduler = new MockScheduler(time);
        Agent agent = createAgent(scheduler);
        AgentClient client = new AgentClient.Builder().
            maxTries(10).target("localhost", agent.port()).build();
        AgentStatusResponse status = client.status();
        assertEquals(Collections.emptyMap(), status.workers());
        new ExpectedTasks().waitFor(client);

        final NoOpTaskSpec fooSpec = new NoOpTaskSpec(1000, 600000);
        client.createWorker(new CreateWorkerRequest(0, "foo", fooSpec));
        new ExpectedTasks().addTask(new ExpectedTaskBuilder("foo").
                workerState(new WorkerRunning("foo", fooSpec, 0, new TextNode("active"))).
                build()).
            waitFor(client);

        final NoOpTaskSpec barSpec = new NoOpTaskSpec(2000, 900000);
        client.createWorker(new CreateWorkerRequest(1, "bar", barSpec));
        client.createWorker(new CreateWorkerRequest(1, "bar", barSpec));

        assertThrows(RequestConflictException.class,
            () -> client.createWorker(new CreateWorkerRequest(1, "foo", barSpec)),
            "Recreating a request with a different taskId is not allowed");
        assertThrows(RequestConflictException.class,
            () -> client.createWorker(new CreateWorkerRequest(1, "bar", fooSpec)),
            "Recreating a request with a different spec is not allowed");

        new ExpectedTasks().
            addTask(new ExpectedTaskBuilder("foo").
                workerState(new WorkerRunning("foo", fooSpec, 0, new TextNode("active"))).
                build()).
            addTask(new ExpectedTaskBuilder("bar").
                workerState(new WorkerRunning("bar", barSpec, 0, new TextNode("active"))).
                build()).
            waitFor(client);

        final NoOpTaskSpec bazSpec = new NoOpTaskSpec(1, 450000);
        client.createWorker(new CreateWorkerRequest(2, "baz", bazSpec));
        new ExpectedTasks().
            addTask(new ExpectedTaskBuilder("foo").
                workerState(new WorkerRunning("foo", fooSpec, 0, new TextNode("active"))).
                build()).
            addTask(new ExpectedTaskBuilder("bar").
                workerState(new WorkerRunning("bar", barSpec, 0, new TextNode("active"))).
                build()).
            addTask(new ExpectedTaskBuilder("baz").
                workerState(new WorkerRunning("baz", bazSpec, 0, new TextNode("active"))).
                build()).
            waitFor(client);
    }
Modify sleep() to call super.sleep() every second call
in order to avoid the endless loop in the tick() calls to the MockScheduler listener
trogdor/src/test/java/org/apache/kafka/trogdor/agent/AgentTest.java