private void validateSinkConnectorGroupId(Map<String, String> config, Map<String, ConfigValue> validatedConfig) {
        String overriddenConsumerGroupIdConfig = CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + GROUP_ID_CONFIG;
        if (config.containsKey(overriddenConsumerGroupIdConfig)) {
            String consumerGroupId = config.get(overriddenConsumerGroupIdConfig);
            ConfigValue validatedGroupId = validatedConfig.computeIfAbsent(
                    overriddenConsumerGroupIdConfig,
                    p -> new ConfigValue(overriddenConsumerGroupIdConfig, consumerGroupId, Collections.emptyList(), new ArrayList<>())
            );
            if (workerGroupId.equals(consumerGroupId)) {
                validatedGroupId.addErrorMessage("Consumer group " + consumerGroupId +
                        " conflicts with Connect worker group " + workerGroupId);
            }
        } else {
            ConfigValue validatedName = validatedConfig.get(ConnectorConfig.NAME_CONFIG);
            String name = (String) validatedName.value();
            if (workerGroupId.equals(SinkUtils.consumerGroupId(name))) {
                validatedName.addErrorMessage("Consumer group for sink connector named " + name +
                        " conflicts with Connect worker group " + workerGroupId);
            }
        }
    }
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