@Override
    public void deleteConnectorConfig(final String connName, final Callback<Created<ConnectorInfo>> callback) {
        addRequest(
            () -> {
                log.trace("Handling connector config request {}", connName);
                if (!isLeader()) {
                    callback.onCompletion(new NotLeaderException("Only the leader can delete connector configs.", leaderUrl()), null);
                    return null;
                }

                if (!configState.contains(connName)) {
                    callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null);
                } else {
                    log.trace("Removing connector config {} {}", connName, configState.connectors());
                    writeToConfigTopicAsLeader(
                            "removing the config for connector " + connName + " from the config topic",
                            () -> configBackingStore.removeConnectorConfig(connName)
                    );
                    callback.onCompletion(null, new Created<>(false, 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