@Override
    public List<ConfigKeyInfo> connectorPluginConfig(String pluginName) {
        Plugins p = plugins();
        Class<?> pluginClass;
        try {
            pluginClass = p.pluginClass(pluginName);
        } catch (ClassNotFoundException cnfe) {
            throw new NotFoundException("Unknown plugin " + pluginName + ".");
        }

        try (LoaderSwap loaderSwap = p.withClassLoader(pluginClass.getClassLoader())) {
            Object plugin = p.newPlugin(pluginName);
            // Contains definitions coming from Connect framework
            ConfigDef baseConfigDefs = null;
            // Contains definitions specifically declared on the plugin
            ConfigDef pluginConfigDefs;
            if (plugin instanceof SinkConnector) {
                baseConfigDefs = SinkConnectorConfig.configDef();
                pluginConfigDefs = ((SinkConnector) plugin).config();
            } else if (plugin instanceof SourceConnector) {
                baseConfigDefs = SourceConnectorConfig.configDef();
                pluginConfigDefs = ((SourceConnector) plugin).config();
            } else if (plugin instanceof Converter) {
                pluginConfigDefs = ((Converter) plugin).config();
            } else if (plugin instanceof HeaderConverter) {
                pluginConfigDefs = ((HeaderConverter) plugin).config();
            } else if (plugin instanceof Transformation) {
                pluginConfigDefs = ((Transformation<?>) plugin).config();
            } else if (plugin instanceof Predicate) {
                pluginConfigDefs = ((Predicate<?>) plugin).config();
            } else {
                throw new BadRequestException("Invalid plugin class " + pluginName + ". Valid types are sink, source, converter, header_converter, transformation, predicate.");
            }

            // Track config properties by name and, if the same property is defined in multiple places,
            // give precedence to the one defined by the plugin class
            // Preserve the ordering of properties as they're returned from each ConfigDef
            Map<String, ConfigKey> configsMap = new LinkedHashMap<>(pluginConfigDefs.configKeys());
            if (baseConfigDefs != null)
                baseConfigDefs.configKeys().forEach(configsMap::putIfAbsent);

            List<ConfigKeyInfo> results = new ArrayList<>();
            for (ConfigKey configKey : configsMap.values()) {
                results.add(AbstractHerder.convertConfigKey(configKey));
            }
            return results;
        } catch (ClassNotFoundException e) {
            throw new ConnectException("Failed to load plugin class or one of its dependencies", e);
        }
    }
Checks a given {@link ConfigInfos} for validation error messages and adds an exception
to the given {@link Callback} if any were found.

@param configInfos configInfos to read Errors from
@param callback callback to add config error exception to
@return true if errors were found in the config
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java