private void doPutConnectorConfig(
            String connName,
            Map<String, String> config,
            TargetState targetState, boolean allowReplace,
            Callback<Created<ConnectorInfo>> callback) {
        validateConnectorConfig(config, callback.chainStaging((error, configInfos) -> {
            if (error != null) {
                callback.onCompletion(error, null);
                return;
            }

            // Complete the connector config write via another herder request in order to
            // perform the write to the backing store (or forward to the leader) during
            // the "external request" portion of the tick loop
            addRequest(
                    () -> {
                        if (maybeAddConfigErrors(configInfos, callback)) {
                            return null;
                        }

                        log.trace("Handling connector config request {}", connName);
                        if (!isLeader()) {
                            callback.onCompletion(new NotLeaderException("Only the leader can set connector configs.", leaderUrl()), null);
                            return null;
                        }
                        boolean exists = configState.contains(connName);
                        if (!allowReplace && exists) {
                            callback.onCompletion(new AlreadyExistsException("Connector " + connName + " already exists"), null);
                            return null;
                        }

                        log.trace("Submitting connector config {} {} {}", connName, allowReplace, configState.connectors());
                        writeToConfigTopicAsLeader(
                                "writing a config for connector " + connName + " to the config topic",
                                () -> configBackingStore.putConnectorConfig(connName, config, targetState)
                        );

                        // Note that we use the updated connector config despite the fact that we don't have an updated
                        // snapshot yet. The existing task info should still be accurate.
                        ConnectorInfo info = new ConnectorInfo(connName, config, configState.tasks(connName),
                                connectorType(config));
                        callback.onCompletion(null, new Created<>(!exists, info));
                        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