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();
}
```
And we could 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 `Plugins` would still be fairly simple:
```java
public DefinedConfigPlugin newDefinedConfigPlugin(String classOrAlias)
throws ClassNotFoundException {
Class<? extends DefinedConfigPlugin> klass =
pluginClass(delegatingLoader, classOrAlias, DefinedConfigPlugin.class);
return newPlugin(klass);
}
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
