@Override
    public void restartTask(final ConnectorTaskId id, final Callback<Void> callback) {
        addRequest(
            () -> {
                if (checkRebalanceNeeded(callback))
                    return null;

                if (!configState.connectors().contains(id.connector())) {
                    callback.onCompletion(new NotFoundException("Unknown connector: " + id.connector()), null);
                    return null;
                }

                if (configState.taskConfig(id) == null) {
                    callback.onCompletion(new NotFoundException("Unknown task: " + id), null);
                    return null;
                }

                if (assignment.tasks().contains(id)) {
                    try (TickThreadStage stage = new TickThreadStage("restarting task " + id)) {
                        worker.stopAndAwaitTask(id);
                        if (startTask(id))
                            callback.onCompletion(null, null);
                        else
                            callback.onCompletion(new ConnectException("Failed to start task: " + id), null);
                    } catch (Throwable t) {
                        callback.onCompletion(t, null);
                    }
                } else if (isLeader()) {
                    callback.onCompletion(new NotAssignedException("Cannot restart task since it is not assigned to this member", member.ownerUrl(id)), null);
                } else {
                    callback.onCompletion(new NotLeaderException("Cannot restart task since it is not assigned to this member", leaderUrl()), null);
                }
                return null;
            },
            forwardErrorAndTickThreadStages(callback));
    }
@Test
    public void testRestartTask() throws Exception {
        // get the initial assignment
        when(member.memberId()).thenReturn("leader");
        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);
        when(herder.connectorType(anyMap())).thenReturn(ConnectorType.SOURCE);
        expectRebalance(1, Collections.emptyList(), singletonList(TASK0), true);
        expectConfigRefreshAndSnapshot(SNAPSHOT);

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

        when(worker.startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any())).thenReturn(true);

        herder.tick();

        // now handle the task restart
        expectMemberEnsureActive();
        doNothing().when(worker).stopAndAwaitTask(TASK0);
        FutureCallback<Void> callback = new FutureCallback<>();
        herder.restartTask(TASK0, callback);
        herder.tick();
        callback.get(1000L, TimeUnit.MILLISECONDS);

        verify(worker, times(2)).startSourceTask(eq(TASK0), any(), any(), any(), eq(herder), any());
        verify(worker).stopAndAwaitTask(TASK0);
        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
    }