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
