@Override
    public void restartConnector(final String connName, final Callback<Void> callback) {
        restartConnector(0, connName, callback);
    }
@Test
    public void testRestartConnector() throws Exception {

        // get the initial assignment
        when(member.memberId()).thenReturn("leader");
        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
        when(worker.isSinkConnector(CONN1)).thenReturn(Boolean.TRUE);
        expectRebalance(1, singletonList(CONN1), Collections.emptyList(), true);
        expectConfigRefreshAndSnapshot(SNAPSHOT);

        when(statusBackingStore.connectors()).thenReturn(Collections.emptySet());
        expectMemberPoll();

        ArgumentCaptor<Callback<TargetState>> onStart = ArgumentCaptor.forClass(Callback.class);
        doAnswer(invocation -> {
            onStart.getValue().onCompletion(null, TargetState.STARTED);
            return true;
        }).when(worker).startConnector(eq(CONN1), any(), any(), eq(herder), eq(TargetState.STARTED), onStart.capture());
        expectExecuteTaskReconfiguration(true, conn1SinkConfig, invocation -> TASK_CONFIGS);

        // Initial rebalance where this member becomes the leader
        herder.tick();

        expectMemberEnsureActive();

        doNothing().when(worker).stopAndAwaitConnector(CONN1);

        FutureCallback<Void> callback = new FutureCallback<>();
        herder.restartConnector(CONN1, callback);
        herder.tick();
        callback.get(1000L, TimeUnit.MILLISECONDS);

        verify(worker, times(2)).startConnector(eq(CONN1), any(), any(), eq(herder), eq(TargetState.STARTED), any());
        verify(worker, times(2)).connectorTaskConfigs(eq(CONN1), any());
        verify(worker).stopAndAwaitConnector(CONN1);
        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
    }