@Test
public void testRestartTask() throws Exception {

    EasyMock.expect(worker.connectorTaskConfigs(CONN1, MAX_TASKS, null)).andStubReturn(TASK_CONFIGS);
    // get the initial assignment
    EasyMock.expect(member.memberId()).andStubReturn("leader");
    expectRebalance(1, Collections.<String>emptyList(), Collections.singletonList(TASK0));
    expectPostRebalanceCatchup(SNAPSHOT);
    member.poll(EasyMock.anyInt());
    PowerMock.expectLastCall();
    worker.startTask(EasyMock.eq(TASK0), EasyMock.<TaskConfig>anyObject(), EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED));
    PowerMock.expectLastCall();
    // now handle the task restart
    member.wakeup();
    PowerMock.expectLastCall();
    member.ensureActive();
    PowerMock.expectLastCall();
    member.poll(EasyMock.anyInt());
    PowerMock.expectLastCall();
    EasyMock.expect(worker.ownsTask(TASK0)).andReturn(true);
    worker.stopAndAwaitTask(TASK0);
    PowerMock.expectLastCall();
    worker.startTask(EasyMock.eq(TASK0), EasyMock.<TaskConfig>anyObject(), EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED));
    PowerMock.expectLastCall();
    PowerMock.replayAll();
    herder.tick();
    FutureCallback<Void> callback = new FutureCallback<>();
    herder.restartTask(TASK0, callback);
    herder.tick();
    callback.get(1000L, TimeUnit.MILLISECONDS);
    PowerMock.verifyAll();
}