C0urante commented on a change in pull request #11572: URL: https://github.com/apache/kafka/pull/11572#discussion_r814469917
########## File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResource.java ########## @@ -56,16 +69,49 @@ private static final String ALIAS_SUFFIX = "Connector"; private final Herder herder; private final List<ConnectorPluginInfo> connectorPlugins; + private final Map<String, PluginType> pluginsByType; - private static final List<Class<? extends Connector>> CONNECTOR_EXCLUDES = Arrays.asList( + static final List<Class<? extends Connector>> CONNECTOR_EXCLUDES = Arrays.asList( VerifiableSourceConnector.class, VerifiableSinkConnector.class, MockConnector.class, MockSourceConnector.class, MockSinkConnector.class, SchemaSourceConnector.class ); + @SuppressWarnings("rawtypes") + static final List<Class<? extends Transformation>> TRANSFORM_EXCLUDES = Arrays.asList( + PredicatedTransformation.class + ); + public ConnectorPluginsResource(Herder herder) { this.herder = herder; this.connectorPlugins = new ArrayList<>(); + this.pluginsByType = new HashMap<>(); + + // TODO: improve once plugins are allowed to be added/removed during runtime. + for (PluginDesc<Connector> plugin : herder.plugins().connectors()) { + if (!CONNECTOR_EXCLUDES.contains(plugin.pluginClass())) { + connectorPlugins.add(new ConnectorPluginInfo(plugin, PluginType.from(plugin.pluginClass()))); + pluginsByType.put(getAlias(plugin.className()), PluginType.from(plugin.pluginClass())); + } + } + for (PluginDesc<Transformation<?>> transform : herder.plugins().transformations()) { + if (!TRANSFORM_EXCLUDES.contains(transform.pluginClass())) { + connectorPlugins.add(new ConnectorPluginInfo(transform, PluginType.TRANSFORMATION)); + pluginsByType.put(getAlias(transform.className()), PluginType.TRANSFORMATION); + } + } + for (PluginDesc<Predicate<?>> predicate : herder.plugins().predicates()) { + connectorPlugins.add(new ConnectorPluginInfo(predicate, PluginType.PREDICATE)); + pluginsByType.put(getAlias(predicate.className()), PluginType.PREDICATE); + } + for (PluginDesc<Converter> converter : herder.plugins().converters()) { + connectorPlugins.add(new ConnectorPluginInfo(converter, PluginType.CONVERTER)); + pluginsByType.put(getAlias(converter.className()), PluginType.CONVERTER); + } + for (PluginDesc<HeaderConverter> headerConverter : herder.plugins().headerConverters()) { + connectorPlugins.add(new ConnectorPluginInfo(headerConverter, PluginType.HEADER_CONVERTER)); + pluginsByType.put(getAlias(headerConverter.className()), PluginType.HEADER_CONVERTER); + } Review comment: It seems like we're duplicating some of the logic contained in `Plugins` into this class by tracking class alias names and pre-computing plugin type based on them. Did you consider a `Herder` method that only accepted the name of the plugin, and took on the responsibility of deducing the plugin type itself? ```java List<ConfigKeyInfo> connectorPluginConfig(String pluginName); ``` In `AbstractHerder`, we could do something like this: ```java @Override public List<ConfigKeyInfo> connectorPluginConfig(String pluginName) { try { Object plugin = Plugins.newPlugin(pluginName); PluginType pluginType = PluginType.from(plugin.class); List<ConfigKeyInfo> results = new ArrayList<>(); ConfigDef configDefs; switch (pluginType) { case SINK: case SOURCE: configDefs = ((Connector) plugin).config(); break; case CONVERTER: configDefs = ((Converter) plugin).config(); break; // ... Rest of switch statement follows same pattern, and rest of the method remains unchanged } ``` And in `Plugins` we could do this: ```java public Object newPlugin(String classOrAlias) throws ClassNotFoundException { Class<? extends Object> klass = pluginClass(delegatingLoader, classOrAlias, Object.class); return newPlugin(klass); } ``` Or alternatively, we could introduce a common interface for plugins that expose a `ConfigDef`: ```java interface DefinedConfigPlugin { ConfigDef config(); } ``` Which could really simplify some of the `AbstractHerder` logic: ```java @Override public List<ConfigKeyInfo> connectorPluginConfig(String pluginName) { try { DefinedConfigPlugin plugin = Plugins.newDefinedConfigPlugin(pluginName); ConfigDef configDefs = plugin.config(); // No switch statement on plugin type necessary // ... Rest of the method remains unchanged } ``` And the change to `Plugins` would be lightweight as well: ```java public DefinedConfigPlugin newDefinedConfigPlugin(String classOrAlias) throws ClassNotFoundException { Class<? extends DefinedConfigPlugin> klass = pluginClass(delegatingLoader, classOrAlias, DefinedConfigPlugin.class); return newPlugin(klass); } ``` Worth noting that if we want to differentiate to users between "this plugin is not on the worker" and "we don't expose config information for this type of plugin", we'd have to make a few further tweaks. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org