@Override
    public void patchConnectorConfig(String connName, Map<String, String> configPatch, Callback<Created<ConnectorInfo>> callback) {
        log.trace("Submitting connector config patch request {}", connName);
        addRequest(() -> {
            // This reduces (but not completely eliminates) the chance for race conditions.
            if (!isLeader()) {
                callback.onCompletion(new NotLeaderException("Only the leader can set connector configs.", leaderUrl()), null);
                return null;
            }

            ConnectorInfo connectorInfo = connectorInfo(connName);
            if (connectorInfo == null) {
                callback.onCompletion(new NotFoundException("Connector " + connName + " not found", null), null);
            } else {
                Map<String, String> patchedConfig = ConnectUtils.patchConfig(connectorInfo.config(), configPatch);
                doPutConnectorConfig(connName, patchedConfig, null, true, callback);
            }
            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