@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);
        }
    }
@Test
    @SuppressWarnings({"rawtypes", "unchecked"})
    public void testGetConnectorConfigDefWithInvalidPluginType() throws Exception {
        String connName = "AnotherPlugin";
        AbstractHerder herder = testHerder();
        when(worker.getPlugins()).thenReturn(plugins);
        when(plugins.pluginClass(anyString())).thenReturn((Class) Object.class);
        when(plugins.newPlugin(anyString())).thenReturn(new DirectoryConfigProvider());
        assertThrows(BadRequestException.class, () -> herder.connectorPluginConfig(connName));
    }