@Override
    public void stopConnector(final String connName, final Callback<Void> callback) {
        log.trace("Submitting request to transition connector {} to STOPPED state", connName);

        addRequest(
                () -> {
                    if (!configState.contains(connName))
                        throw new NotFoundException("Unknown connector " + connName);

                    // We only allow the leader to handle this request since it involves writing task configs to the config topic
                    if (!isLeader()) {
                        callback.onCompletion(new NotLeaderException("Only the leader can transition connectors to the STOPPED state.", leaderUrl()), null);
                        return null;
                    }

                    // We write the task configs first since, if we fail between then and writing the target state, the
                    // cluster is still kept in a healthy state. A RUNNING connector with zero tasks is acceptable (although,
                    // if the connector is reassigned during the ensuing rebalance, it is likely that it will immediately generate
                    // a non-empty set of task configs). A STOPPED connector with a non-empty set of tasks is less acceptable
                    // and likely to confuse users.
                    writeTaskConfigs(connName, Collections.emptyList());
                    String stageDescription = "writing the STOPPED target stage for connector " + connName + " to the config topic";
                    try (TickThreadStage stage = new TickThreadStage(stageDescription)) {
                        configBackingStore.putTargetState(connName, TargetState.STOPPED);
                    }
                    // Force a read of the new target state for the connector
                    if (!refreshConfigSnapshot(workerSyncTimeoutMs)) {
                        log.warn("Failed to read to end of config topic after writing the STOPPED target state for connector {}", connName);
                    }

                    callback.onCompletion(null, null);
                    return null;
                },
                forwardErrorAndTickThreadStages(callback)
        );
    }
@Test
    public void testStopConnector() throws Exception {
        when(herder.connectorType(anyMap())).thenReturn(ConnectorType.SOURCE);
        when(member.memberId()).thenReturn("leader");
        when(member.currentProtocolVersion()).thenReturn(CONNECT_PROTOCOL_V0);

        // join as leader
        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), eq(TargetState.STARTED))).thenReturn(true);

        herder.tick(); // join

        // handle stop request
        expectMemberEnsureActive();
        expectConfigRefreshAndSnapshot(SNAPSHOT);
        doNothing().when(configBackingStore).putTaskConfigs(CONN1, Collections.emptyList());
        doNothing().when(configBackingStore).putTargetState(CONN1, TargetState.STOPPED);

        FutureCallback<Void> cb = new FutureCallback<>();

        herder.stopConnector(CONN1, cb); // external request
        herder.tick(); // continue

        assertTrue(cb.isDone(), "Callback should already have been invoked by herder");
        cb.get(0, TimeUnit.MILLISECONDS);

        verifyNoMoreInteractions(worker, member, configBackingStore, statusBackingStore);
    }