@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)
        );
    }
Close resources managed by this herder but which are not explicitly started.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java