C0urante commented on a change in pull request #11572:
URL: https://github.com/apache/kafka/pull/11572#discussion_r814358703
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java
##########
@@ -168,6 +173,44 @@ public Connector newConnector(String
connectorClassOrAlias) {
return newPlugin(klass);
}
+ public Converter newConverter(String className) throws
ClassNotFoundException {
Review comment:
Nit: Might be nice to have (brief) Javadocs on these methods. I was
thinking "don't we already use the `Plugins` class to instantiate converters?
Why do we need another?". After taking a look it's clear now that this is for
just instantiating a converter in order to grab its `ConfigDef`, and the other
methods (like `Plugins.newConverter(AbstractConfig config, String
classPropertyName, ClassLoaderUsage classLoaderUsage)`) are for instantiating
and configuring converters for use with tasks.
##########
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:
Also, could we abstract and simplify the logic for populating
`connectorPlugins`? It looks like the only wrinkle here preventing us from
writing a generic `private void addConnectorPlugins(Collection<PluginDesc<?>>,
PluginType pluginType)` method and then invoking that once each for sink
connectors, source connectors, converters, etc. is that the `Plugins` class
doesn't expose source and sink connectors separately.
Considering that `Plugins::connectors` is only used (excluding test code) in
one place--this class--could we refactor that into two separate
`Plugins::sinkConnectors` and `Plugins::sourceConnectors` methods?
##########
File path:
connect/api/src/main/java/org/apache/kafka/connect/storage/Converter.java
##########
@@ -86,4 +87,12 @@
default SchemaAndValue toConnectData(String topic, Headers headers, byte[]
value) {
return toConnectData(topic, value);
}
+
+ /**
+ * Configuration specification for this set of converters.
Review comment:
I know this matches the Javadoc from the KIP but I'm wondering why we're
using "converters" plural here? Would it be more clear to say "for this
converter" instead?
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
##########
@@ -613,7 +614,7 @@ public static ConfigInfos generateResult(String connType,
Map<String, ConfigKey>
return new ConfigInfos(connType, errorCount, groups, configInfoList);
}
- private static ConfigKeyInfo convertConfigKey(ConfigKey configKey) {
+ public static ConfigKeyInfo convertConfigKey(ConfigKey configKey) {
Review comment:
Nit: kind of strange that this method (and honestly a lot of the config
wrangling logic) is contained in the `AbstractHerder` class. Might be worth
refactoring into a separate config utils class/package. Probably best to leave
for a separate PR, but if you agree that that'd be cleaner I could see about
doing that work as a follow-up.
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java
##########
@@ -168,6 +173,44 @@ public Connector newConnector(String
connectorClassOrAlias) {
return newPlugin(klass);
}
+ public Converter newConverter(String className) throws
ClassNotFoundException {
+ Class<? extends Converter> klass = pluginClass(
+ delegatingLoader,
+ className,
+ Converter.class
+ );
+ return newPlugin(klass);
+ }
+
+ public HeaderConverter newHeaderConverter(String className) throws
ClassNotFoundException {
+ Class<? extends HeaderConverter> klass = pluginClass(
+ delegatingLoader,
+ className,
+ HeaderConverter.class
+ );
+ return newPlugin(klass);
+ }
+
+ @SuppressWarnings("rawtypes")
+ public Predicate newPredicate(String className) throws
ClassNotFoundException {
Review comment:
This should match the type-safe declaration for `newTransformation` IMO:
```suggestion
@SuppressWarnings({"unchecked", "rawtypes"})
public <R extends ConnectRecord<R>> Predicate<R> newPredicate(String
className) throws ClassNotFoundException {
```
I think this is due to testing woes with the
`AbstractHerderTest::testConnectorPluginConfig` test case; I've left a
suggestion in that section for how to keep the type safety here with a small
change there.
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorPluginInfo.java
##########
@@ -79,11 +82,21 @@ public int hashCode() {
@Override
public String toString() {
- final StringBuilder sb = new StringBuilder("ConnectorPluginInfo{");
- sb.append("className='").append(className).append('\'');
- sb.append(", type=").append(type);
- sb.append(", version='").append(version).append('\'');
- sb.append('}');
- return sb.toString();
+ return "ConnectorPluginInfo{" + "className='" + className + '\'' +
+ ", type=" + type.toString() +
+ ", version='" + version + '\'' +
+ '}';
+ }
+
+ public static final class NoVersionFilter {
+ @Override
+ public boolean equals(Object obj) {
+ return DelegatingClassLoader.UNDEFINED_VERSION.equals(obj);
+ }
+
+ @Override
+ public int hashCode() {
+ return super.hashCode();
+ }
Review comment:
Do we need this method declaration at all? Looks like it might have been
left in accidentally while trying to wrestle with SpotBugs?
```suggestion
```
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
##########
@@ -750,4 +751,38 @@ private String trace(Throwable t) {
return keys;
}
+ @Override
Review comment:
(Left thoughts in `ConnectorPluginsResource` about possible ways we
could refactor this to reduce the workload on the REST layer to handle some of
the classloading logic)
##########
File path:
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java
##########
@@ -901,6 +902,48 @@ public void
testGenerateResultWithConfigValuesWithNoConfigKeysAndWithSomeErrors(
assertInfoValue(infos, "config.extra2", "value.extra2", "error
extra2");
}
+ @Test
+ public void testConnectorPluginConfig() throws Exception {
+ AbstractHerder herder = partialMockBuilder(AbstractHerder.class)
+ .withConstructor(
+ Worker.class,
+ String.class,
+ String.class,
+ StatusBackingStore.class,
+ ConfigBackingStore.class,
+ ConnectorClientConfigOverridePolicy.class
+ )
+ .withArgs(worker, workerId, kafkaClusterId, statusStore,
configStore, noneConnectorClientConfigOverridePolicy)
+ .addMockedMethod("generation")
+ .createMock();
+
+
EasyMock.expect(plugins.newConnector(EasyMock.anyString())).andReturn(new
SampleSourceConnector()).anyTimes();
+
EasyMock.expect(plugins.newConverter(EasyMock.anyString())).andReturn(new
SampleConverterWithHeaders()).anyTimes();
+
EasyMock.expect(plugins.newHeaderConverter(EasyMock.anyString())).andReturn(new
SampleHeaderConverter()).anyTimes();
+
EasyMock.expect(plugins.newPredicate(EasyMock.anyString())).andReturn(new
SamplePredicate()).anyTimes();
Review comment:
If we want to keep the type-safe signature for `Plugins::newPredicate`:
```suggestion
EasyMock.expect(plugins.<SourceRecord>newPredicate(EasyMock.anyString())).andReturn(new
SamplePredicate()).anyTimes();
```
Note that this also requires adding an import for
`org.apache.kafka.connect.source.SourceRecord` to this test class.
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResource.java
##########
@@ -123,4 +171,15 @@ private String normalizedPluginName(String pluginName) {
? pluginName.substring(0, pluginName.length() -
ALIAS_SUFFIX.length())
: pluginName;
}
+
+ String getAlias(String name) {
+ name = normalizedPluginName(name);
+ int lastIndexOf = name.lastIndexOf('.');
+ return lastIndexOf >= 0 ? name.substring(lastIndexOf + 1) : name;
+ }
+
+ private synchronized List<ConfigKeyInfo> doGetConfigDef(final String
pluginName) {
+ PluginType pluginType =
pluginsByType.getOrDefault(getAlias(pluginName), PluginType.UNKNOWN);
Review comment:
Also, it seems like using a default of `PluginType.UNKNOWN` here might
be suboptimal. If someone wants to the view the config for a REST extension,
for example, they'll end up seeing an error message later on (in
`AbstractHerder::connectorPluginConfig`) that says something like "Invalid
plugin type unknown. Valid types are..."
I think it'd be clearer to users if we could differentiate between these two
cases:
1. User requests config for a plugin that does exist on the worker, but
which we don't expose config information via the REST API for (such as a REST
extension or a config provider)
2. User requests config for a plugin that doesn't exist on the worker
Status-wise, In the case of 1, a 400 response probably makes sense, but for
2, a 404 response might be more applicable.
##########
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();
}
```
(this could be kept package-private so as not to quality as public interface)
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);
}
```
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorPluginInfo.java
##########
@@ -17,30 +17,32 @@
package org.apache.kafka.connect.runtime.rest.entities;
import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
-import org.apache.kafka.connect.connector.Connector;
+import org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader;
import org.apache.kafka.connect.runtime.isolation.PluginDesc;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
import java.util.Objects;
public class ConnectorPluginInfo {
private final String className;
- private final ConnectorType type;
+ private final PluginType type;
private final String version;
@JsonCreator
public ConnectorPluginInfo(
@JsonProperty("class") String className,
- @JsonProperty("type") ConnectorType type,
+ @JsonProperty("type") PluginType type,
@JsonProperty("version") String version
) {
this.className = className;
this.type = type;
this.version = version;
}
- public ConnectorPluginInfo(PluginDesc<Connector> plugin) {
- this(plugin.className(), ConnectorType.from(plugin.pluginClass()),
plugin.version());
+ public ConnectorPluginInfo(PluginDesc<?> plugin, PluginType type) {
+ this(plugin.className(), type, plugin.version());
Review comment:
Isn't the `PluginType` argument redundant if we already have a
`PluginDesc`?
```suggestion
public ConnectorPluginInfo(PluginDesc<?> plugin) {
this(plugin.className(), plugin.type(), plugin.version());
```
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResource.java
##########
@@ -123,4 +171,15 @@ private String normalizedPluginName(String pluginName) {
? pluginName.substring(0, pluginName.length() -
ALIAS_SUFFIX.length())
: pluginName;
}
+
+ String getAlias(String name) {
+ name = normalizedPluginName(name);
+ int lastIndexOf = name.lastIndexOf('.');
+ return lastIndexOf >= 0 ? name.substring(lastIndexOf + 1) : name;
+ }
+
+ private synchronized List<ConfigKeyInfo> doGetConfigDef(final String
pluginName) {
+ PluginType pluginType =
pluginsByType.getOrDefault(getAlias(pluginName), PluginType.UNKNOWN);
Review comment:
Doesn't this introduce the possibility of conflict between two plugins
(or I guess specifically connectors, since those are the only ones we strip
suffixes from) which have different fully-qualified class names, but the same
simple class name? Or where they would have the same simple class name, except
that one ends with `Connector` and the other doesn't?
In practice this is unlikely to come up but if we support it at the moment,
probably best to take care here to avoid introducing a potential regression,
especially if someone for some reason wants to run, e.g., two different
`MySqlSink` connectors on their worker.
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResource.java
##########
@@ -100,21 +146,23 @@ public ConfigInfos validateConfigs(
@GET
@Path("/")
- public List<ConnectorPluginInfo> listConnectorPlugins() {
- return getConnectorPlugins();
+ public List<ConnectorPluginInfo>
listConnectorPlugins(@DefaultValue("true") @QueryParam("connectorsOnly")
boolean connectorsOnly) {
+ return getConnectorPlugins(connectorsOnly);
Review comment:
Nit: I know this is following the [existing
style](https://github.com/apache/kafka/blob/c2ee1411c8bb73fcf96c12abeedbfe6fde2c6354/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResource.java#L101-L118)
in the code base, but do you think the separate `getConnectorPlugins` method
is actually bringing anything to the table readability-wise?
Think we could just as easily eliminate the `getConnectorPlugins` method and
inline it directly here. Same thought with `getConnectorConfigDef`.
--
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]
