@Test
public void testRestartConnector() 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.singletonList(CONN1), Collections.<ConnectorTaskId>emptyList());
    expectPostRebalanceCatchup(SNAPSHOT);
    member.poll(EasyMock.anyInt());
    PowerMock.expectLastCall();
    worker.startConnector(EasyMock.<ConnectorConfig>anyObject(), EasyMock.<ConnectorContext>anyObject(),
            EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED));
    PowerMock.expectLastCall();
    EasyMock.expect(worker.isRunning(CONN1)).andReturn(true);
    // now handle the connector restart
    member.wakeup();
    PowerMock.expectLastCall();
    member.ensureActive();
    PowerMock.expectLastCall();
    member.poll(EasyMock.anyInt());
    PowerMock.expectLastCall();
    EasyMock.expect(worker.ownsConnector(CONN1)).andReturn(true);
    worker.stopConnector(CONN1);
    PowerMock.expectLastCall();
    worker.startConnector(EasyMock.<ConnectorConfig>anyObject(), EasyMock.<ConnectorContext>anyObject(),
            EasyMock.eq(herder), EasyMock.eq(TargetState.STARTED));
    PowerMock.expectLastCall();
    EasyMock.expect(worker.isRunning(CONN1)).andReturn(true);
    PowerMock.replayAll();
    herder.tick();
    FutureCallback<Void> callback = new FutureCallback<>();
    herder.restartConnector(CONN1, callback);
    herder.tick();
    callback.get(1000L, TimeUnit.MILLISECONDS);
    PowerMock.verifyAll();
}