You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mi...@apache.org on 2022/03/03 13:30:43 UTC
[kafka] branch trunk updated: KAFKA-13510: Connect APIs to list all connector plugins and retrieve their configs (#11572)
This is an automated email from the ASF dual-hosted git repository.
mimaison pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 029a14b KAFKA-13510: Connect APIs to list all connector plugins and retrieve their configs (#11572)
029a14b is described below
commit 029a14b530dcaab980faee4fa18c925b95db3122
Author: Mickael Maison <mi...@users.noreply.github.com>
AuthorDate: Thu Mar 3 14:28:50 2022 +0100
KAFKA-13510: Connect APIs to list all connector plugins and retrieve their configs (#11572)
Implements KIP-769: https://cwiki.apache.org/confluence/display/KAFKA/KIP-769%3A+Connect+APIs+to+list+all+connector+plugins+and+retrieve+their+configuration+definitions
Reviewers: Tom Bentley <tb...@redhat.com>, Chris Egerton <fe...@gmail.com>
---
.../apache/kafka/connect/storage/Converter.java | 9 +
.../kafka/connect/runtime/AbstractHerder.java | 44 ++-
.../org/apache/kafka/connect/runtime/Herder.java | 9 +
.../connect/runtime/PredicatedTransformation.java | 2 +-
.../runtime/isolation/DelegatingClassLoader.java | 46 ++-
.../runtime/isolation/PluginScanResult.java | 22 +-
.../connect/runtime/isolation/PluginType.java | 6 +-
.../connect/runtime/isolation/PluginUtils.java | 1 -
.../kafka/connect/runtime/isolation/Plugins.java | 33 +-
.../{ConnectorPluginInfo.java => PluginInfo.java} | 48 +--
.../rest/resources/ConnectorPluginsResource.java | 75 +++--
.../connect/tools/VerifiableSinkConnector.java | 4 +-
.../integration/MonitorableSinkConnector.java | 4 +-
.../integration/MonitorableSourceConnector.java | 4 +-
.../kafka/connect/runtime/AbstractHerderTest.java | 186 +++++++-----
.../runtime/PredicatedTransformationTest.java | 79 +----
...eaders.java => SampleConverterWithHeaders.java} | 2 +-
...nkConnector.java => SampleHeaderConverter.java} | 38 +--
...TestSinkConnector.java => SamplePredicate.java} | 40 ++-
...SinkConnector.java => SampleSinkConnector.java} | 2 +-
...ceConnector.java => SampleSourceConnector.java} | 2 +-
...inkConnector.java => SampleTransformation.java} | 40 ++-
.../kafka/connect/runtime/WorkerSinkTaskTest.java | 2 +-
.../connect/runtime/WorkerSourceTaskTest.java | 2 +-
.../runtime/rest/entities/PluginInfoTest.java | 35 +++
.../resources/ConnectorPluginsResourceTest.java | 333 +++++++++++----------
gradle/spotbugs-exclude.xml | 8 +
27 files changed, 621 insertions(+), 455 deletions(-)
diff --git a/connect/api/src/main/java/org/apache/kafka/connect/storage/Converter.java b/connect/api/src/main/java/org/apache/kafka/connect/storage/Converter.java
index 2d2ef4a..51b23be 100644
--- a/connect/api/src/main/java/org/apache/kafka/connect/storage/Converter.java
+++ b/connect/api/src/main/java/org/apache/kafka/connect/storage/Converter.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.connect.storage;
+import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaAndValue;
@@ -86,4 +87,12 @@ public interface Converter {
default SchemaAndValue toConnectData(String topic, Headers headers, byte[] value) {
return toConnectData(topic, value);
}
+
+ /**
+ * Configuration specification for this converter.
+ * @return the configuration specification; may not be null
+ */
+ default ConfigDef config() {
+ return new ConfigDef();
+ }
}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
index 0fc8e2b..2fe75a9 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
@@ -29,6 +29,7 @@ import org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePo
import org.apache.kafka.connect.connector.policy.ConnectorClientConfigRequest;
import org.apache.kafka.connect.errors.NotFoundException;
import org.apache.kafka.connect.runtime.distributed.ClusterConfigState;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
import org.apache.kafka.connect.runtime.isolation.Plugins;
import org.apache.kafka.connect.runtime.rest.entities.ActiveTopicsInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfo;
@@ -41,7 +42,11 @@ import org.apache.kafka.connect.runtime.rest.entities.ConnectorType;
import org.apache.kafka.connect.runtime.rest.errors.BadRequestException;
import org.apache.kafka.connect.source.SourceConnector;
import org.apache.kafka.connect.storage.ConfigBackingStore;
+import org.apache.kafka.connect.storage.Converter;
+import org.apache.kafka.connect.storage.HeaderConverter;
import org.apache.kafka.connect.storage.StatusBackingStore;
+import org.apache.kafka.connect.transforms.Transformation;
+import org.apache.kafka.connect.transforms.predicates.Predicate;
import org.apache.kafka.connect.util.Callback;
import org.apache.kafka.connect.util.ConnectorTaskId;
@@ -613,7 +618,7 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
return new ConfigInfos(connType, errorCount, groups, configInfoList);
}
- private static ConfigKeyInfo convertConfigKey(ConfigKey configKey) {
+ public static ConfigKeyInfo convertConfigKey(ConfigKey configKey) {
return convertConfigKey(configKey, "");
}
@@ -750,4 +755,41 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
return keys;
}
+ @Override
+ public List<ConfigKeyInfo> connectorPluginConfig(String pluginName) {
+ List<ConfigKeyInfo> results = new ArrayList<>();
+ ConfigDef configDefs;
+ try {
+ Plugins p = plugins();
+ Object plugin = p.newPlugin(pluginName);
+ PluginType pluginType = PluginType.from(plugin.getClass());
+ switch (pluginType) {
+ case SINK:
+ case SOURCE:
+ configDefs = ((Connector) plugin).config();
+ break;
+ case CONVERTER:
+ configDefs = ((Converter) plugin).config();
+ break;
+ case HEADER_CONVERTER:
+ configDefs = ((HeaderConverter) plugin).config();
+ break;
+ case TRANSFORMATION:
+ configDefs = ((Transformation<?>) plugin).config();
+ break;
+ case PREDICATE:
+ configDefs = ((Predicate<?>) plugin).config();
+ break;
+ default:
+ throw new BadRequestException("Invalid plugin type " + pluginType + ". Valid types are sink, source, converter, header_converter, transformation, predicate.");
+ }
+ } catch (ClassNotFoundException cnfe) {
+ throw new NotFoundException("Unknown plugin " + pluginName + ".");
+ }
+ for (ConfigDef.ConfigKey configKey : configDefs.configKeys().values()) {
+ results.add(AbstractHerder.convertConfigKey(configKey));
+ }
+ return results;
+ }
+
}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java
index 945797f..0f20c0b 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java
@@ -20,6 +20,7 @@ import org.apache.kafka.connect.runtime.isolation.Plugins;
import org.apache.kafka.connect.runtime.rest.InternalRequestSignature;
import org.apache.kafka.connect.runtime.rest.entities.ActiveTopicsInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos;
+import org.apache.kafka.connect.runtime.rest.entities.ConfigKeyInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
import org.apache.kafka.connect.runtime.rest.entities.TaskInfo;
@@ -258,6 +259,14 @@ public interface Herder {
*/
String kafkaClusterId();
+
+ /**
+ * Returns the configuration of a plugin
+ * @param pluginName the name of the plugin
+ * @return the list of ConfigKeyInfo of the plugin
+ */
+ List<ConfigKeyInfo> connectorPluginConfig(String pluginName);
+
enum ConfigReloadAction {
NONE,
RESTART
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/PredicatedTransformation.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/PredicatedTransformation.java
index 0af0e72..8b7db30 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/PredicatedTransformation.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/PredicatedTransformation.java
@@ -30,7 +30,7 @@ import org.apache.kafka.connect.transforms.predicates.Predicate;
* {@link Predicate} is true (or false, according to {@code negate}).
* @param <R>
*/
-class PredicatedTransformation<R extends ConnectRecord<R>> implements Transformation<R> {
+public class PredicatedTransformation<R extends ConnectRecord<R>> implements Transformation<R> {
static final String PREDICATE_CONFIG = "predicate";
static final String NEGATE_CONFIG = "negate";
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java
index cf5a548..b4f4b88 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java
@@ -21,6 +21,8 @@ import org.apache.kafka.connect.components.Versioned;
import org.apache.kafka.connect.connector.Connector;
import org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.rest.ConnectRestExtension;
+import org.apache.kafka.connect.sink.SinkConnector;
+import org.apache.kafka.connect.source.SourceConnector;
import org.apache.kafka.connect.storage.Converter;
import org.apache.kafka.connect.storage.HeaderConverter;
import org.apache.kafka.connect.transforms.Transformation;
@@ -76,11 +78,12 @@ import java.util.stream.Collectors;
public class DelegatingClassLoader extends URLClassLoader {
private static final Logger log = LoggerFactory.getLogger(DelegatingClassLoader.class);
private static final String CLASSPATH_NAME = "classpath";
- private static final String UNDEFINED_VERSION = "undefined";
+ public static final String UNDEFINED_VERSION = "undefined";
private final ConcurrentMap<String, SortedMap<PluginDesc<?>, ClassLoader>> pluginLoaders;
private final ConcurrentMap<String, String> aliases;
- private final SortedSet<PluginDesc<Connector>> connectors;
+ private final SortedSet<PluginDesc<SinkConnector>> sinkConnectors;
+ private final SortedSet<PluginDesc<SourceConnector>> sourceConnectors;
private final SortedSet<PluginDesc<Converter>> converters;
private final SortedSet<PluginDesc<HeaderConverter>> headerConverters;
private final SortedSet<PluginDesc<Transformation<?>>> transformations;
@@ -109,7 +112,8 @@ public class DelegatingClassLoader extends URLClassLoader {
this.pluginPaths = pluginPaths;
this.pluginLoaders = new ConcurrentHashMap<>();
this.aliases = new ConcurrentHashMap<>();
- this.connectors = new TreeSet<>();
+ this.sinkConnectors = new TreeSet<>();
+ this.sourceConnectors = new TreeSet<>();
this.converters = new TreeSet<>();
this.headerConverters = new TreeSet<>();
this.transformations = new TreeSet<>();
@@ -127,10 +131,21 @@ public class DelegatingClassLoader extends URLClassLoader {
this(pluginPaths, DelegatingClassLoader.class.getClassLoader());
}
+ @SuppressWarnings({"unchecked", "rawtypes"})
public Set<PluginDesc<Connector>> connectors() {
+ Set<PluginDesc<Connector>> connectors = new TreeSet<>((Set) sinkConnectors);
+ connectors.addAll((Set) sourceConnectors);
return connectors;
}
+ public Set<PluginDesc<SinkConnector>> sinkConnectors() {
+ return sinkConnectors;
+ }
+
+ public Set<PluginDesc<SourceConnector>> sourceConnectors() {
+ return sourceConnectors;
+ }
+
public Set<PluginDesc<Converter>> converters() {
return converters;
}
@@ -232,8 +247,7 @@ public class DelegatingClassLoader extends URLClassLoader {
if (CLASSPATH_NAME.equals(path)) {
scanUrlsAndAddPlugins(
getParent(),
- ClasspathHelper.forJavaClassPath().toArray(new URL[0]),
- null
+ ClasspathHelper.forJavaClassPath().toArray(new URL[0])
);
} else {
Path pluginPath = Paths.get(path).toAbsolutePath();
@@ -254,7 +268,7 @@ public class DelegatingClassLoader extends URLClassLoader {
} catch (IOException e) {
log.error("Could not get listing for plugin path: {}. Ignoring.", path, e);
} catch (ReflectiveOperationException e) {
- log.error("Could not instantiate plugins in: {}. Ignoring: {}", path, e);
+ log.error("Could not instantiate plugins in: {}. Ignoring.", path, e);
}
}
@@ -274,19 +288,20 @@ public class DelegatingClassLoader extends URLClassLoader {
urls,
this
);
- scanUrlsAndAddPlugins(loader, urls, pluginLocation);
+ scanUrlsAndAddPlugins(loader, urls);
}
private void scanUrlsAndAddPlugins(
ClassLoader loader,
- URL[] urls,
- Path pluginLocation
+ URL[] urls
) throws ReflectiveOperationException {
PluginScanResult plugins = scanPluginPath(loader, urls);
log.info("Registered loader: {}", loader);
if (!plugins.isEmpty()) {
- addPlugins(plugins.connectors(), loader);
- connectors.addAll(plugins.connectors());
+ addPlugins(plugins.sinkConnectors(), loader);
+ sinkConnectors.addAll(plugins.sinkConnectors());
+ addPlugins(plugins.sourceConnectors(), loader);
+ sourceConnectors.addAll(plugins.sourceConnectors());
addPlugins(plugins.converters(), loader);
converters.addAll(plugins.converters());
addPlugins(plugins.headerConverters(), loader);
@@ -348,7 +363,8 @@ public class DelegatingClassLoader extends URLClassLoader {
Reflections reflections = new InternalReflections(builder);
return new PluginScanResult(
- getPluginDesc(reflections, Connector.class, loader),
+ getPluginDesc(reflections, SinkConnector.class, loader),
+ getPluginDesc(reflections, SourceConnector.class, loader),
getPluginDesc(reflections, Converter.class, loader),
getPluginDesc(reflections, HeaderConverter.class, loader),
getTransformationPluginDesc(loader, reflections),
@@ -419,7 +435,7 @@ public class DelegatingClassLoader extends URLClassLoader {
return pluginImpl instanceof Versioned ? ((Versioned) pluginImpl).version() : UNDEFINED_VERSION;
}
- private static <T> String versionFor(Class<? extends T> pluginKlass) throws ReflectiveOperationException {
+ public static <T> String versionFor(Class<? extends T> pluginKlass) throws ReflectiveOperationException {
// Temporary workaround until all the plugins are versioned.
return Connector.class.isAssignableFrom(pluginKlass) ?
versionFor(pluginKlass.getDeclaredConstructor().newInstance()) : UNDEFINED_VERSION;
@@ -438,7 +454,7 @@ public class DelegatingClassLoader extends URLClassLoader {
}
private void addAllAliases() {
- addAliases(connectors);
+ addAliases(connectors());
addAliases(converters);
addAliases(headerConverters);
addAliases(transformations);
@@ -503,7 +519,7 @@ public class DelegatingClassLoader extends URLClassLoader {
@Override
public Enumeration<URL> getResources(String name) throws IOException {
if (serviceLoaderManifestForPlugin(name)) {
- // Default implementation of getResources searches the parent class loader and and also its own URL paths. This will enable the
+ // Default implementation of getResources searches the parent class loader and also its own URL paths. This will enable the
// PluginClassLoader to limit its resource search to only its own URL paths.
return null;
} else {
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginScanResult.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginScanResult.java
index e98945e..565fe6d 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginScanResult.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginScanResult.java
@@ -17,9 +17,10 @@
package org.apache.kafka.connect.runtime.isolation;
import org.apache.kafka.common.config.provider.ConfigProvider;
-import org.apache.kafka.connect.connector.Connector;
import org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.rest.ConnectRestExtension;
+import org.apache.kafka.connect.sink.SinkConnector;
+import org.apache.kafka.connect.source.SourceConnector;
import org.apache.kafka.connect.storage.Converter;
import org.apache.kafka.connect.storage.HeaderConverter;
import org.apache.kafka.connect.transforms.Transformation;
@@ -30,7 +31,8 @@ import java.util.Collection;
import java.util.List;
public class PluginScanResult {
- private final Collection<PluginDesc<Connector>> connectors;
+ private final Collection<PluginDesc<SinkConnector>> sinkConnectors;
+ private final Collection<PluginDesc<SourceConnector>> sourceConnectors;
private final Collection<PluginDesc<Converter>> converters;
private final Collection<PluginDesc<HeaderConverter>> headerConverters;
private final Collection<PluginDesc<Transformation<?>>> transformations;
@@ -42,7 +44,8 @@ public class PluginScanResult {
private final List<Collection<?>> allPlugins;
public PluginScanResult(
- Collection<PluginDesc<Connector>> connectors,
+ Collection<PluginDesc<SinkConnector>> sinkConnectors,
+ Collection<PluginDesc<SourceConnector>> sourceConnectors,
Collection<PluginDesc<Converter>> converters,
Collection<PluginDesc<HeaderConverter>> headerConverters,
Collection<PluginDesc<Transformation<?>>> transformations,
@@ -51,7 +54,8 @@ public class PluginScanResult {
Collection<PluginDesc<ConnectRestExtension>> restExtensions,
Collection<PluginDesc<ConnectorClientConfigOverridePolicy>> connectorClientConfigPolicies
) {
- this.connectors = connectors;
+ this.sinkConnectors = sinkConnectors;
+ this.sourceConnectors = sourceConnectors;
this.converters = converters;
this.headerConverters = headerConverters;
this.transformations = transformations;
@@ -60,12 +64,16 @@ public class PluginScanResult {
this.restExtensions = restExtensions;
this.connectorClientConfigPolicies = connectorClientConfigPolicies;
this.allPlugins =
- Arrays.asList(connectors, converters, headerConverters, transformations, configProviders,
+ Arrays.asList(sinkConnectors, sourceConnectors, converters, headerConverters, transformations, configProviders,
connectorClientConfigPolicies);
}
- public Collection<PluginDesc<Connector>> connectors() {
- return connectors;
+ public Collection<PluginDesc<SinkConnector>> sinkConnectors() {
+ return sinkConnectors;
+ }
+
+ public Collection<PluginDesc<SourceConnector>> sourceConnectors() {
+ return sourceConnectors;
}
public Collection<PluginDesc<Converter>> converters() {
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginType.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginType.java
index c26b180..696e14b 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginType.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginType.java
@@ -17,22 +17,24 @@
package org.apache.kafka.connect.runtime.isolation;
import org.apache.kafka.common.config.provider.ConfigProvider;
-import org.apache.kafka.connect.connector.Connector;
import org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.rest.ConnectRestExtension;
import org.apache.kafka.connect.sink.SinkConnector;
import org.apache.kafka.connect.source.SourceConnector;
import org.apache.kafka.connect.storage.Converter;
+import org.apache.kafka.connect.storage.HeaderConverter;
import org.apache.kafka.connect.transforms.Transformation;
+import org.apache.kafka.connect.transforms.predicates.Predicate;
import java.util.Locale;
public enum PluginType {
SOURCE(SourceConnector.class),
SINK(SinkConnector.class),
- CONNECTOR(Connector.class),
CONVERTER(Converter.class),
+ HEADER_CONVERTER(HeaderConverter.class),
TRANSFORMATION(Transformation.class),
+ PREDICATE(Predicate.class),
CONFIGPROVIDER(ConfigProvider.class),
REST_EXTENSION(ConnectRestExtension.class),
CONNECTOR_CLIENT_CONFIG_OVERRIDE_POLICY(ConnectorClientConfigOverridePolicy.class),
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginUtils.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginUtils.java
index 12cb1867..46a4cfa 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginUtils.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginUtils.java
@@ -319,7 +319,6 @@ public class PluginUtils {
switch (plugin.type()) {
case SOURCE:
case SINK:
- case CONNECTOR:
return prunePluginName(plugin, "Connector");
default:
return prunePluginName(plugin, plugin.type().simpleName());
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java
index 26b5d50..7ec73ba 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java
@@ -25,6 +25,8 @@ import org.apache.kafka.connect.connector.Connector;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.runtime.WorkerConfig;
+import org.apache.kafka.connect.sink.SinkConnector;
+import org.apache.kafka.connect.source.SourceConnector;
import org.apache.kafka.connect.storage.Converter;
import org.apache.kafka.connect.storage.ConverterConfig;
import org.apache.kafka.connect.storage.ConverterType;
@@ -147,14 +149,22 @@ public class Plugins {
return delegatingLoader;
}
- public Set<PluginDesc<Connector>> connectors() {
- return delegatingLoader.connectors();
+ public Set<PluginDesc<SinkConnector>> sinkConnectors() {
+ return delegatingLoader.sinkConnectors();
+ }
+
+ public Set<PluginDesc<SourceConnector>> sourceConnectors() {
+ return delegatingLoader.sourceConnectors();
}
public Set<PluginDesc<Converter>> converters() {
return delegatingLoader.converters();
}
+ public Set<PluginDesc<HeaderConverter>> headerConverters() {
+ return delegatingLoader.headerConverters();
+ }
+
public Set<PluginDesc<Transformation<?>>> transformations() {
return delegatingLoader.transformations();
}
@@ -163,6 +173,11 @@ public class Plugins {
return delegatingLoader.predicates();
}
+ public Object newPlugin(String classOrAlias) throws ClassNotFoundException {
+ Class<?> klass = pluginClass(delegatingLoader, classOrAlias, Object.class);
+ return newPlugin(klass);
+ }
+
public Connector newConnector(String connectorClassOrAlias) {
Class<? extends Connector> klass = connectorClass(connectorClassOrAlias);
return newPlugin(klass);
@@ -177,8 +192,9 @@ public class Plugins {
Connector.class
);
} catch (ClassNotFoundException e) {
- List<PluginDesc<Connector>> matches = new ArrayList<>();
- for (PluginDesc<Connector> plugin : delegatingLoader.connectors()) {
+ List<PluginDesc<? extends Connector>> matches = new ArrayList<>();
+ Set<PluginDesc<Connector>> connectors = delegatingLoader.connectors();
+ for (PluginDesc<? extends Connector> plugin : connectors) {
Class<?> pluginClass = plugin.pluginClass();
String simpleName = pluginClass.getSimpleName();
if (simpleName.equals(connectorClassOrAlias)
@@ -192,20 +208,19 @@ public class Plugins {
"Failed to find any class that implements Connector and which name matches "
+ connectorClassOrAlias
+ ", available connectors are: "
- + pluginNames(delegatingLoader.connectors())
+ + Utils.join(connectors, ", ")
);
}
if (matches.size() > 1) {
throw new ConnectException(
"More than one connector matches alias "
+ connectorClassOrAlias
- +
- ". Please use full package and class name instead. Classes found: "
- + pluginNames(matches)
+ + ". Please use full package and class name instead. Classes found: "
+ + Utils.join(connectors, ", ")
);
}
- PluginDesc<Connector> entry = matches.get(0);
+ PluginDesc<? extends Connector> entry = matches.get(0);
klass = entry.pluginClass();
}
return klass;
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorPluginInfo.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/PluginInfo.java
similarity index 60%
rename from connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorPluginInfo.java
rename to connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/PluginInfo.java
index 98b30e1..e703267 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorPluginInfo.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/PluginInfo.java
@@ -17,21 +17,23 @@
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 {
+public class PluginInfo {
private final String className;
- private final ConnectorType type;
+ private final PluginType type;
private final String version;
@JsonCreator
- public ConnectorPluginInfo(
+ public PluginInfo(
@JsonProperty("class") String className,
- @JsonProperty("type") ConnectorType type,
+ @JsonProperty("type") PluginType type,
@JsonProperty("version") String version
) {
this.className = className;
@@ -39,8 +41,8 @@ public class ConnectorPluginInfo {
this.version = version;
}
- public ConnectorPluginInfo(PluginDesc<Connector> plugin) {
- this(plugin.className(), ConnectorType.from(plugin.pluginClass()), plugin.version());
+ public PluginInfo(PluginDesc<?> plugin) {
+ this(plugin.className(), plugin.type(), plugin.version());
}
@JsonProperty("class")
@@ -49,11 +51,12 @@ public class ConnectorPluginInfo {
}
@JsonProperty("type")
- public ConnectorType type() {
- return type;
+ public String type() {
+ return type.toString();
}
@JsonProperty("version")
+ @JsonInclude(value = JsonInclude.Include.CUSTOM, valueFilter = NoVersionFilter.class)
public String version() {
return version;
}
@@ -66,9 +69,9 @@ public class ConnectorPluginInfo {
if (o == null || getClass() != o.getClass()) {
return false;
}
- ConnectorPluginInfo that = (ConnectorPluginInfo) o;
+ PluginInfo that = (PluginInfo) o;
return Objects.equals(className, that.className) &&
- type == that.type &&
+ Objects.equals(type, that.type) &&
Objects.equals(version, that.version);
}
@@ -79,11 +82,22 @@ public class ConnectorPluginInfo {
@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 "PluginInfo{" + "className='" + className + '\'' +
+ ", type=" + type.toString() +
+ ", version='" + version + '\'' +
+ '}';
+ }
+
+ public static final class NoVersionFilter {
+ // This method is used by Jackson to filter the version field for plugins that don't have a version
+ public boolean equals(Object obj) {
+ return DelegatingClassLoader.UNDEFINED_VERSION.equals(obj);
+ }
+
+ // Dummy hashCode method to not fail compilation because of equals() method
+ @Override
+ public int hashCode() {
+ return super.hashCode();
+ }
}
}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResource.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResource.java
index 0854c8f..2beda9f 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResource.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResource.java
@@ -16,37 +16,45 @@
*/
package org.apache.kafka.connect.runtime.rest.resources;
-import org.apache.kafka.connect.connector.Connector;
import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.Herder;
+import org.apache.kafka.connect.runtime.PredicatedTransformation;
import org.apache.kafka.connect.runtime.isolation.PluginDesc;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos;
-import org.apache.kafka.connect.runtime.rest.entities.ConnectorPluginInfo;
+import org.apache.kafka.connect.runtime.rest.entities.ConfigKeyInfo;
+import org.apache.kafka.connect.runtime.rest.entities.PluginInfo;
import org.apache.kafka.connect.runtime.rest.errors.ConnectRestException;
-import org.apache.kafka.connect.tools.MockConnector;
+import org.apache.kafka.connect.sink.SinkConnector;
+import org.apache.kafka.connect.source.SourceConnector;
import org.apache.kafka.connect.tools.MockSinkConnector;
import org.apache.kafka.connect.tools.MockSourceConnector;
import org.apache.kafka.connect.tools.SchemaSourceConnector;
import org.apache.kafka.connect.tools.VerifiableSinkConnector;
import org.apache.kafka.connect.tools.VerifiableSourceConnector;
+import org.apache.kafka.connect.transforms.Transformation;
import org.apache.kafka.connect.util.FutureCallback;
import javax.ws.rs.BadRequestException;
import javax.ws.rs.Consumes;
+import javax.ws.rs.DefaultValue;
import javax.ws.rs.GET;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
@Path("/connector-plugins")
@Produces(MediaType.APPLICATION_JSON)
@@ -55,17 +63,42 @@ public class ConnectorPluginsResource {
private static final String ALIAS_SUFFIX = "Connector";
private final Herder herder;
- private final List<ConnectorPluginInfo> connectorPlugins;
+ private final List<PluginInfo> connectorPlugins;
- private static final List<Class<? extends Connector>> CONNECTOR_EXCLUDES = Arrays.asList(
- VerifiableSourceConnector.class, VerifiableSinkConnector.class,
- MockConnector.class, MockSourceConnector.class, MockSinkConnector.class,
+ static final List<Class<? extends SinkConnector>> SINK_CONNECTOR_EXCLUDES = Arrays.asList(
+ VerifiableSinkConnector.class,
+ MockSinkConnector.class
+ );
+
+ static final List<Class<? extends SourceConnector>> SOURCE_CONNECTOR_EXCLUDES = Arrays.asList(
+ VerifiableSourceConnector.class,
+ MockSourceConnector.class,
SchemaSourceConnector.class
);
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ static final List<Class<? extends Transformation<?>>> TRANSFORM_EXCLUDES = Collections.singletonList(
+ (Class) PredicatedTransformation.class
+ );
+
public ConnectorPluginsResource(Herder herder) {
this.herder = herder;
this.connectorPlugins = new ArrayList<>();
+
+ // TODO: improve once plugins are allowed to be added/removed during runtime.
+ addConnectorPlugins(herder.plugins().sinkConnectors(), SINK_CONNECTOR_EXCLUDES);
+ addConnectorPlugins(herder.plugins().sourceConnectors(), SOURCE_CONNECTOR_EXCLUDES);
+ addConnectorPlugins(herder.plugins().transformations(), TRANSFORM_EXCLUDES);
+ addConnectorPlugins(herder.plugins().predicates(), Collections.emptySet());
+ addConnectorPlugins(herder.plugins().converters(), Collections.emptySet());
+ addConnectorPlugins(herder.plugins().headerConverters(), Collections.emptySet());
+ }
+
+ private <T> void addConnectorPlugins(Collection<PluginDesc<T>> plugins, Collection<Class<? extends T>> excludes) {
+ plugins.stream()
+ .filter(p -> !excludes.contains(p.pluginClass()))
+ .map(PluginInfo::new)
+ .forEach(connectorPlugins::add);
}
@PUT
@@ -100,21 +133,24 @@ public class ConnectorPluginsResource {
@GET
@Path("/")
- public List<ConnectorPluginInfo> listConnectorPlugins() {
- return getConnectorPlugins();
- }
-
- // TODO: improve once plugins are allowed to be added/removed during runtime.
- private synchronized List<ConnectorPluginInfo> getConnectorPlugins() {
- if (connectorPlugins.isEmpty()) {
- for (PluginDesc<Connector> plugin : herder.plugins().connectors()) {
- if (!CONNECTOR_EXCLUDES.contains(plugin.pluginClass())) {
- connectorPlugins.add(new ConnectorPluginInfo(plugin));
- }
+ public List<PluginInfo> listConnectorPlugins(@DefaultValue("true") @QueryParam("connectorsOnly") boolean connectorsOnly) {
+ synchronized (this) {
+ if (connectorsOnly) {
+ return Collections.unmodifiableList(connectorPlugins.stream()
+ .filter(p -> PluginType.SINK.toString().equals(p.type()) || PluginType.SOURCE.toString().equals(p.type()))
+ .collect(Collectors.toList()));
+ } else {
+ return Collections.unmodifiableList(connectorPlugins);
}
}
+ }
- return Collections.unmodifiableList(connectorPlugins);
+ @GET
+ @Path("/{name}/config")
+ public List<ConfigKeyInfo> getConnectorConfigDef(final @PathParam("name") String pluginName) {
+ synchronized (this) {
+ return herder.connectorPluginConfig(pluginName);
+ }
}
private String normalizedPluginName(String pluginName) {
@@ -123,4 +159,5 @@ public class ConnectorPluginsResource {
? pluginName.substring(0, pluginName.length() - ALIAS_SUFFIX.length())
: pluginName;
}
+
}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/tools/VerifiableSinkConnector.java b/connect/runtime/src/main/java/org/apache/kafka/connect/tools/VerifiableSinkConnector.java
index 55f95a3..fbe29e1 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/tools/VerifiableSinkConnector.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/tools/VerifiableSinkConnector.java
@@ -19,7 +19,7 @@ package org.apache.kafka.connect.tools;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.connect.connector.Task;
-import org.apache.kafka.connect.source.SourceConnector;
+import org.apache.kafka.connect.sink.SinkConnector;
import java.util.ArrayList;
import java.util.HashMap;
@@ -29,7 +29,7 @@ import java.util.Map;
/**
* @see VerifiableSinkTask
*/
-public class VerifiableSinkConnector extends SourceConnector {
+public class VerifiableSinkConnector extends SinkConnector {
private Map<String, String> config;
@Override
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkConnector.java
index 5733199..8309d72 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkConnector.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkConnector.java
@@ -20,7 +20,7 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.connector.Task;
-import org.apache.kafka.connect.runtime.TestSinkConnector;
+import org.apache.kafka.connect.runtime.SampleSinkConnector;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTask;
import org.slf4j.Logger;
@@ -39,7 +39,7 @@ import java.util.Map;
* which are initiated by the embedded connector, and wait for them to consume a desired number of
* messages.
*/
-public class MonitorableSinkConnector extends TestSinkConnector {
+public class MonitorableSinkConnector extends SampleSinkConnector {
private static final Logger log = LoggerFactory.getLogger(MonitorableSinkConnector.class);
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java
index 390927c..4f13ad0 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java
@@ -21,7 +21,7 @@ import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.header.ConnectHeaders;
-import org.apache.kafka.connect.runtime.TestSourceConnector;
+import org.apache.kafka.connect.runtime.SampleSourceConnector;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
import org.apache.kafka.tools.ThroughputThrottler;
@@ -43,7 +43,7 @@ import java.util.stream.LongStream;
* that generates records of a fixed structure. The rate of record production can be adjusted
* through the configs 'throughput' and 'messages.per.poll'
*/
-public class MonitorableSourceConnector extends TestSourceConnector {
+public class MonitorableSourceConnector extends SampleSourceConnector {
private static final Logger log = LoggerFactory.getLogger(MonitorableSourceConnector.class);
public static final String TOPIC_CONFIG = "topic";
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java
index d64dbaa..5b9e199 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java
@@ -23,18 +23,20 @@ import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.ConfigTransformer;
import org.apache.kafka.common.config.ConfigValue;
import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.provider.DirectoryConfigProvider;
import org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredLoginCallbackHandler;
-import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.connector.Connector;
import org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.connector.policy.NoneConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.connector.policy.PrincipalConnectorClientConfigOverridePolicy;
+import org.apache.kafka.connect.errors.NotFoundException;
import org.apache.kafka.connect.runtime.distributed.ClusterConfigState;
import org.apache.kafka.connect.runtime.isolation.PluginDesc;
import org.apache.kafka.connect.runtime.isolation.Plugins;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos;
+import org.apache.kafka.connect.runtime.rest.entities.ConfigKeyInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConfigValueInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorType;
@@ -69,6 +71,7 @@ import java.util.function.Function;
import java.util.stream.Collectors;
import static org.apache.kafka.connect.runtime.AbstractHerder.keysWithVariableValues;
+import static org.easymock.EasyMock.anyString;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import static org.powermock.api.easymock.PowerMock.verifyAll;
@@ -292,11 +295,11 @@ public class AbstractHerderTest {
@Test()
public void testConfigValidationNullConfig() {
- AbstractHerder herder = createConfigValidationHerder(TestSourceConnector.class, noneConnectorClientConfigOverridePolicy);
+ AbstractHerder herder = createConfigValidationHerder(SampleSourceConnector.class, noneConnectorClientConfigOverridePolicy);
replayAll();
Map<String, String> config = new HashMap<>();
- config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSourceConnector.class.getName());
+ config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, SampleSourceConnector.class.getName());
config.put("name", "somename");
config.put("required", "value");
config.put("testKey", null);
@@ -311,11 +314,11 @@ public class AbstractHerderTest {
@Test
public void testConfigValidationMultipleNullConfig() {
- AbstractHerder herder = createConfigValidationHerder(TestSourceConnector.class, noneConnectorClientConfigOverridePolicy);
+ AbstractHerder herder = createConfigValidationHerder(SampleSourceConnector.class, noneConnectorClientConfigOverridePolicy);
replayAll();
Map<String, String> config = new HashMap<>();
- config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSourceConnector.class.getName());
+ config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, SampleSourceConnector.class.getName());
config.put("name", "somename");
config.put("required", "value");
config.put("testKey", null);
@@ -414,7 +417,7 @@ public class AbstractHerderTest {
@Test
public void testConfigValidationEmptyConfig() {
- AbstractHerder herder = createConfigValidationHerder(TestSourceConnector.class, noneConnectorClientConfigOverridePolicy, 0);
+ AbstractHerder herder = createConfigValidationHerder(SampleSourceConnector.class, noneConnectorClientConfigOverridePolicy, 0);
replayAll();
assertThrows(BadRequestException.class, () -> herder.validateConnectorConfig(Collections.emptyMap(), false));
@@ -424,15 +427,15 @@ public class AbstractHerderTest {
@Test()
public void testConfigValidationMissingName() {
- AbstractHerder herder = createConfigValidationHerder(TestSourceConnector.class, noneConnectorClientConfigOverridePolicy);
+ AbstractHerder herder = createConfigValidationHerder(SampleSourceConnector.class, noneConnectorClientConfigOverridePolicy);
replayAll();
- Map<String, String> config = Collections.singletonMap(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSourceConnector.class.getName());
+ Map<String, String> config = Collections.singletonMap(ConnectorConfig.CONNECTOR_CLASS_CONFIG, SampleSourceConnector.class.getName());
ConfigInfos result = herder.validateConnectorConfig(config, false);
// We expect there to be errors due to the missing name and .... Note that these assertions depend heavily on
// the config fields for SourceConnectorConfig, but we expect these to change rarely.
- assertEquals(TestSourceConnector.class.getName(), result.name());
+ assertEquals(SampleSourceConnector.class.getName(), result.name());
assertEquals(Arrays.asList(ConnectorConfig.COMMON_GROUP, ConnectorConfig.TRANSFORMS_GROUP,
ConnectorConfig.PREDICATES_GROUP, ConnectorConfig.ERROR_GROUP, SourceConnectorConfig.TOPIC_CREATION_GROUP), result.groups());
assertEquals(2, result.errorCount());
@@ -453,11 +456,11 @@ public class AbstractHerderTest {
@Test
public void testConfigValidationInvalidTopics() {
- AbstractHerder herder = createConfigValidationHerder(TestSinkConnector.class, noneConnectorClientConfigOverridePolicy);
+ AbstractHerder herder = createConfigValidationHerder(SampleSinkConnector.class, noneConnectorClientConfigOverridePolicy);
replayAll();
Map<String, String> config = new HashMap<>();
- config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSinkConnector.class.getName());
+ config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, SampleSinkConnector.class.getName());
config.put(SinkConnectorConfig.TOPICS_CONFIG, "topic1,topic2");
config.put(SinkConnectorConfig.TOPICS_REGEX_CONFIG, "topic.*");
@@ -468,11 +471,11 @@ public class AbstractHerderTest {
@Test
public void testConfigValidationTopicsWithDlq() {
- AbstractHerder herder = createConfigValidationHerder(TestSinkConnector.class, noneConnectorClientConfigOverridePolicy);
+ AbstractHerder herder = createConfigValidationHerder(SampleSinkConnector.class, noneConnectorClientConfigOverridePolicy);
replayAll();
Map<String, String> config = new HashMap<>();
- config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSinkConnector.class.getName());
+ config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, SampleSinkConnector.class.getName());
config.put(SinkConnectorConfig.TOPICS_CONFIG, "topic1");
config.put(SinkConnectorConfig.DLQ_TOPIC_NAME_CONFIG, "topic1");
@@ -483,11 +486,11 @@ public class AbstractHerderTest {
@Test
public void testConfigValidationTopicsRegexWithDlq() {
- AbstractHerder herder = createConfigValidationHerder(TestSinkConnector.class, noneConnectorClientConfigOverridePolicy);
+ AbstractHerder herder = createConfigValidationHerder(SampleSinkConnector.class, noneConnectorClientConfigOverridePolicy);
replayAll();
Map<String, String> config = new HashMap<>();
- config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSinkConnector.class.getName());
+ config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, SampleSinkConnector.class.getName());
config.put(SinkConnectorConfig.TOPICS_REGEX_CONFIG, "topic.*");
config.put(SinkConnectorConfig.DLQ_TOPIC_NAME_CONFIG, "topic1");
@@ -498,7 +501,7 @@ public class AbstractHerderTest {
@Test()
public void testConfigValidationTransformsExtendResults() {
- AbstractHerder herder = createConfigValidationHerder(TestSourceConnector.class, noneConnectorClientConfigOverridePolicy);
+ AbstractHerder herder = createConfigValidationHerder(SampleSourceConnector.class, noneConnectorClientConfigOverridePolicy);
// 2 transform aliases defined -> 2 plugin lookups
Set<PluginDesc<Transformation<?>>> transformations = new HashSet<>();
@@ -510,7 +513,7 @@ public class AbstractHerderTest {
// Define 2 transformations. One has a class defined and so can get embedded configs, the other is missing
// class info that should generate an error.
Map<String, String> config = new HashMap<>();
- config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSourceConnector.class.getName());
+ config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, SampleSourceConnector.class.getName());
config.put(ConnectorConfig.NAME_CONFIG, "connector-name");
config.put(ConnectorConfig.TRANSFORMS_CONFIG, "xformA,xformB");
config.put(ConnectorConfig.TRANSFORMS_CONFIG + ".xformA.type", SampleTransformation.class.getName());
@@ -520,7 +523,7 @@ public class AbstractHerderTest {
// We expect there to be errors due to the missing name and .... Note that these assertions depend heavily on
// the config fields for SourceConnectorConfig, but we expect these to change rarely.
- assertEquals(TestSourceConnector.class.getName(), result.name());
+ assertEquals(SampleSourceConnector.class.getName(), result.name());
// Each transform also gets its own group
List<String> expectedGroups = Arrays.asList(
ConnectorConfig.COMMON_GROUP,
@@ -550,7 +553,7 @@ public class AbstractHerderTest {
@Test()
public void testConfigValidationPredicatesExtendResults() {
- AbstractHerder herder = createConfigValidationHerder(TestSourceConnector.class, noneConnectorClientConfigOverridePolicy);
+ AbstractHerder herder = createConfigValidationHerder(SampleSourceConnector.class, noneConnectorClientConfigOverridePolicy);
// 2 transform aliases defined -> 2 plugin lookups
Set<PluginDesc<Transformation<?>>> transformations = new HashSet<>();
@@ -566,7 +569,7 @@ public class AbstractHerderTest {
// Define 2 transformations. One has a class defined and so can get embedded configs, the other is missing
// class info that should generate an error.
Map<String, String> config = new HashMap<>();
- config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSourceConnector.class.getName());
+ config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, SampleSourceConnector.class.getName());
config.put(ConnectorConfig.NAME_CONFIG, "connector-name");
config.put(ConnectorConfig.TRANSFORMS_CONFIG, "xformA");
config.put(ConnectorConfig.TRANSFORMS_CONFIG + ".xformA.type", SampleTransformation.class.getName());
@@ -579,7 +582,7 @@ public class AbstractHerderTest {
// We expect there to be errors due to the missing name and .... Note that these assertions depend heavily on
// the config fields for SourceConnectorConfig, but we expect these to change rarely.
- assertEquals(TestSourceConnector.class.getName(), result.name());
+ assertEquals(SampleSourceConnector.class.getName(), result.name());
// Each transform also gets its own group
List<String> expectedGroups = Arrays.asList(
ConnectorConfig.COMMON_GROUP,
@@ -632,11 +635,11 @@ public class AbstractHerderTest {
@Test()
public void testConfigValidationPrincipalOnlyOverride() {
- AbstractHerder herder = createConfigValidationHerder(TestSourceConnector.class, new PrincipalConnectorClientConfigOverridePolicy());
+ AbstractHerder herder = createConfigValidationHerder(SampleSourceConnector.class, new PrincipalConnectorClientConfigOverridePolicy());
replayAll();
Map<String, String> config = new HashMap<>();
- config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSourceConnector.class.getName());
+ config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, SampleSourceConnector.class.getName());
config.put(ConnectorConfig.NAME_CONFIG, "connector-name");
config.put("required", "value"); // connector required config
String ackConfigKey = producerOverrideKey(ProducerConfig.ACKS_CONFIG);
@@ -649,7 +652,7 @@ public class AbstractHerderTest {
// We expect there to be errors due to now allowed override policy for ACKS.... Note that these assertions depend heavily on
// the config fields for SourceConnectorConfig, but we expect these to change rarely.
- assertEquals(TestSourceConnector.class.getName(), result.name());
+ assertEquals(SampleSourceConnector.class.getName(), result.name());
// Each transform also gets its own group
List<String> expectedGroups = Arrays.asList(
ConnectorConfig.COMMON_GROUP,
@@ -672,11 +675,11 @@ public class AbstractHerderTest {
@Test
public void testConfigValidationAllOverride() {
- AbstractHerder herder = createConfigValidationHerder(TestSourceConnector.class, new AllConnectorClientConfigOverridePolicy());
+ AbstractHerder herder = createConfigValidationHerder(SampleSourceConnector.class, new AllConnectorClientConfigOverridePolicy());
replayAll();
Map<String, String> config = new HashMap<>();
- config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSourceConnector.class.getName());
+ config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, SampleSourceConnector.class.getName());
config.put(ConnectorConfig.NAME_CONFIG, "connector-name");
config.put("required", "value"); // connector required config
// Try to test a variety of configuration types: string, int, long, boolean, list, class
@@ -901,6 +904,90 @@ public class AbstractHerderTest {
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.newPlugin(EasyMock.anyString())).andAnswer(() -> {
+ String name = (String) EasyMock.getCurrentArguments()[0];
+ switch (name) {
+ case "sink": return new SampleSinkConnector();
+ case "source": return new SampleSourceConnector();
+ case "converter": return new SampleConverterWithHeaders();
+ case "header-converter": return new SampleHeaderConverter();
+ case "predicate": return new SamplePredicate();
+ default: return new SampleTransformation<>();
+ }
+ }).anyTimes();
+ EasyMock.expect(herder.plugins()).andStubReturn(plugins);
+ replayAll();
+
+ List<ConfigKeyInfo> sinkConnectorConfigs = herder.connectorPluginConfig("sink");
+ assertNotNull(sinkConnectorConfigs);
+ assertEquals(new SampleSinkConnector().config().names().size(), sinkConnectorConfigs.size());
+
+ List<ConfigKeyInfo> sourceConnectorConfigs = herder.connectorPluginConfig("source");
+ assertNotNull(sourceConnectorConfigs);
+ assertEquals(new SampleSourceConnector().config().names().size(), sourceConnectorConfigs.size());
+
+ List<ConfigKeyInfo> converterConfigs = herder.connectorPluginConfig("converter");
+ assertNotNull(converterConfigs);
+ assertEquals(new SampleConverterWithHeaders().config().names().size(), converterConfigs.size());
+
+ List<ConfigKeyInfo> headerConverterConfigs = herder.connectorPluginConfig("header-converter");
+ assertNotNull(headerConverterConfigs);
+ assertEquals(new SampleHeaderConverter().config().names().size(), headerConverterConfigs.size());
+
+ List<ConfigKeyInfo> predicateConfigs = herder.connectorPluginConfig("predicate");
+ assertNotNull(predicateConfigs);
+ assertEquals(new SamplePredicate().config().names().size(), predicateConfigs.size());
+
+ List<ConfigKeyInfo> transformationConfigs = herder.connectorPluginConfig("transformation");
+ assertNotNull(transformationConfigs);
+ assertEquals(new SampleTransformation<>().config().names().size(), transformationConfigs.size());
+ }
+
+ @Test(expected = NotFoundException.class)
+ public void testGetConnectorConfigDefWithBadName() throws Exception {
+ String connName = "AnotherPlugin";
+ 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(worker.getPlugins()).andStubReturn(plugins);
+ EasyMock.expect(plugins.newPlugin(anyString())).andThrow(new ClassNotFoundException());
+ replayAll();
+ herder.connectorPluginConfig(connName);
+ }
+
+ @Test(expected = BadRequestException.class)
+ public void testGetConnectorConfigDefWithInvalidPluginType() throws Exception {
+ String connName = "AnotherPlugin";
+ 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(worker.getPlugins()).andStubReturn(plugins);
+ EasyMock.expect(plugins.newPlugin(anyString())).andReturn(new DirectoryConfigProvider());
+ replayAll();
+ herder.connectorPluginConfig(connName);
+ }
+
protected void addConfigKey(Map<String, ConfigDef.ConfigKey> keys, String name, String group) {
keys.put(name, new ConfigDef.ConfigKey(name, ConfigDef.Type.STRING, null, null,
ConfigDef.Importance.HIGH, "doc", group, 10,
@@ -987,53 +1074,6 @@ public class AbstractHerderTest {
return herder;
}
- public static class SampleTransformation<R extends ConnectRecord<R>> implements Transformation<R> {
- @Override
- public void configure(Map<String, ?> configs) {
-
- }
-
- @Override
- public R apply(R record) {
- return record;
- }
-
- @Override
- public ConfigDef config() {
- return new ConfigDef()
- .define("subconfig", ConfigDef.Type.STRING, "default", ConfigDef.Importance.LOW, "docs");
- }
-
- @Override
- public void close() {
-
- }
- }
-
- public static class SamplePredicate<R extends ConnectRecord<R>> implements Predicate<R> {
-
- @Override
- public ConfigDef config() {
- return new ConfigDef()
- .define("predconfig", ConfigDef.Type.STRING, "default", ConfigDef.Importance.LOW, "docs");
- }
-
- @Override
- public boolean test(R record) {
- return false;
- }
-
- @Override
- public void close() {
-
- }
-
- @Override
- public void configure(Map<String, ?> configs) {
-
- }
- }
-
// We need to use a real class here due to some issue with mocking java.lang.Class
private abstract class BogusSourceConnector extends SourceConnector {
}
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/PredicatedTransformationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/PredicatedTransformationTest.java
index 75542e7..8bce328 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/PredicatedTransformationTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/PredicatedTransformationTest.java
@@ -16,12 +16,7 @@
*/
package org.apache.kafka.connect.runtime;
-import java.util.Map;
-
-import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.source.SourceRecord;
-import org.apache.kafka.connect.transforms.Transformation;
-import org.apache.kafka.connect.transforms.predicates.Predicate;
import org.junit.Test;
import static java.util.Collections.singletonMap;
@@ -43,75 +38,9 @@ public class PredicatedTransformationTest {
private void applyAndAssert(boolean predicateResult, boolean negate,
SourceRecord expectedResult) {
- class TestTransformation implements Transformation<SourceRecord> {
-
- private boolean closed = false;
- private SourceRecord transformedRecord;
-
- private TestTransformation(SourceRecord transformedRecord) {
- this.transformedRecord = transformedRecord;
- }
-
- @Override
- public SourceRecord apply(SourceRecord record) {
- return transformedRecord;
- }
-
- @Override
- public ConfigDef config() {
- return null;
- }
-
- @Override
- public void close() {
- closed = true;
- }
-
- @Override
- public void configure(Map<String, ?> configs) {
-
- }
-
- private void assertClosed() {
- assertTrue("Transformer should be closed", closed);
- }
- }
-
- class TestPredicate implements Predicate<SourceRecord> {
-
- private boolean testResult;
- private boolean closed = false;
-
- private TestPredicate(boolean testResult) {
- this.testResult = testResult;
- }
-
- @Override
- public ConfigDef config() {
- return null;
- }
-
- @Override
- public boolean test(SourceRecord record) {
- return testResult;
- }
-
- @Override
- public void close() {
- closed = true;
- }
-
- @Override
- public void configure(Map<String, ?> configs) {
-
- }
- private void assertClosed() {
- assertTrue("Predicate should be closed", closed);
- }
- }
- TestPredicate predicate = new TestPredicate(predicateResult);
- TestTransformation predicatedTransform = new TestTransformation(transformed);
+ SamplePredicate predicate = new SamplePredicate(predicateResult);
+ SampleTransformation<SourceRecord> predicatedTransform = new SampleTransformation<>(transformed);
PredicatedTransformation<SourceRecord> pt = new PredicatedTransformation<>(
predicate,
negate,
@@ -120,7 +49,7 @@ public class PredicatedTransformationTest {
assertEquals(expectedResult, pt.apply(initial));
pt.close();
- predicate.assertClosed();
- predicatedTransform.assertClosed();
+ assertTrue(predicate.closed);
+ assertTrue(predicatedTransform.closed);
}
}
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestConverterWithHeaders.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleConverterWithHeaders.java
similarity index 97%
rename from connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestConverterWithHeaders.java
rename to connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleConverterWithHeaders.java
index 91e0999..3927ba7 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestConverterWithHeaders.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleConverterWithHeaders.java
@@ -28,7 +28,7 @@ import org.apache.kafka.connect.storage.Converter;
/**
* This is a simple Converter implementation that uses "encoding" header to encode/decode strings via provided charset name
*/
-public class TestConverterWithHeaders implements Converter {
+public class SampleConverterWithHeaders implements Converter {
private static final String HEADER_ENCODING = "encoding";
@Override
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSinkConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleHeaderConverter.java
similarity index 59%
copy from connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSinkConnector.java
copy to connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleHeaderConverter.java
index a6e3bb1..ed11360 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSinkConnector.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleHeaderConverter.java
@@ -17,45 +17,37 @@
package org.apache.kafka.connect.runtime;
import org.apache.kafka.common.config.ConfigDef;
-import org.apache.kafka.connect.connector.Task;
-import org.apache.kafka.connect.sink.SinkConnector;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaAndValue;
+import org.apache.kafka.connect.storage.HeaderConverter;
-import java.util.List;
+import java.io.IOException;
import java.util.Map;
-public class TestSinkConnector extends SinkConnector {
-
- public static final String VERSION = "some great version";
-
+public class SampleHeaderConverter implements HeaderConverter {
@Override
- public String version() {
- return VERSION;
- }
-
- @Override
- public void start(Map<String, String> props) {
-
+ public SchemaAndValue toConnectHeader(String topic, String headerKey, byte[] value) {
+ return null;
}
@Override
- public Class<? extends Task> taskClass() {
- return null;
+ public byte[] fromConnectHeader(String topic, String headerKey, Schema schema, Object value) {
+ return new byte[0];
}
@Override
- public List<Map<String, String>> taskConfigs(int maxTasks) {
- return null;
+ public ConfigDef config() {
+ return new ConfigDef()
+ .define("converterconfig", ConfigDef.Type.STRING, "default", ConfigDef.Importance.LOW, "docs");
}
@Override
- public void stop() {
+ public void close() throws IOException {
}
@Override
- public ConfigDef config() {
- return new ConfigDef()
- .define("required", ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "required docs")
- .define("optional", ConfigDef.Type.STRING, "defaultVal", ConfigDef.Importance.HIGH, "optional docs");
+ public void configure(Map<String, ?> configs) {
+
}
}
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSinkConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SamplePredicate.java
similarity index 57%
copy from connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSinkConnector.java
copy to connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SamplePredicate.java
index a6e3bb1..90dfba7 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSinkConnector.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SamplePredicate.java
@@ -17,45 +17,39 @@
package org.apache.kafka.connect.runtime;
import org.apache.kafka.common.config.ConfigDef;
-import org.apache.kafka.connect.connector.Task;
-import org.apache.kafka.connect.sink.SinkConnector;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.apache.kafka.connect.transforms.predicates.Predicate;
-import java.util.List;
import java.util.Map;
-public class TestSinkConnector extends SinkConnector {
+public class SamplePredicate implements Predicate<SourceRecord> {
- public static final String VERSION = "some great version";
+ private boolean testResult;
+ boolean closed = false;
- @Override
- public String version() {
- return VERSION;
- }
-
- @Override
- public void start(Map<String, String> props) {
+ public SamplePredicate() { }
+ public SamplePredicate(boolean testResult) {
+ this.testResult = testResult;
}
@Override
- public Class<? extends Task> taskClass() {
- return null;
+ public ConfigDef config() {
+ return new ConfigDef()
+ .define("predconfig", ConfigDef.Type.STRING, "default", ConfigDef.Importance.LOW, "docs");
}
@Override
- public List<Map<String, String>> taskConfigs(int maxTasks) {
- return null;
+ public boolean test(SourceRecord record) {
+ return testResult;
}
@Override
- public void stop() {
-
+ public void close() {
+ closed = true;
}
@Override
- public ConfigDef config() {
- return new ConfigDef()
- .define("required", ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "required docs")
- .define("optional", ConfigDef.Type.STRING, "defaultVal", ConfigDef.Importance.HIGH, "optional docs");
- }
+ public void configure(Map<String, ?> configs) { }
+
}
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSinkConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleSinkConnector.java
similarity index 96%
copy from connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSinkConnector.java
copy to connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleSinkConnector.java
index a6e3bb1..1c324fe 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSinkConnector.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleSinkConnector.java
@@ -23,7 +23,7 @@ import org.apache.kafka.connect.sink.SinkConnector;
import java.util.List;
import java.util.Map;
-public class TestSinkConnector extends SinkConnector {
+public class SampleSinkConnector extends SinkConnector {
public static final String VERSION = "some great version";
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSourceConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleSourceConnector.java
similarity index 96%
rename from connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSourceConnector.java
rename to connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleSourceConnector.java
index 5f754e2..3615954 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSourceConnector.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleSourceConnector.java
@@ -23,7 +23,7 @@ import org.apache.kafka.connect.source.SourceConnector;
import java.util.List;
import java.util.Map;
-public class TestSourceConnector extends SourceConnector {
+public class SampleSourceConnector extends SourceConnector {
public static final String VERSION = "an entirely different version";
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSinkConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleTransformation.java
similarity index 57%
rename from connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSinkConnector.java
rename to connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleTransformation.java
index a6e3bb1..b904359 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/TestSinkConnector.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleTransformation.java
@@ -17,45 +17,39 @@
package org.apache.kafka.connect.runtime;
import org.apache.kafka.common.config.ConfigDef;
-import org.apache.kafka.connect.connector.Task;
-import org.apache.kafka.connect.sink.SinkConnector;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.transforms.Transformation;
-import java.util.List;
import java.util.Map;
-public class TestSinkConnector extends SinkConnector {
+public class SampleTransformation<R extends ConnectRecord<R>> implements Transformation<R> {
- public static final String VERSION = "some great version";
+ boolean closed = false;
+ private R transformedRecord;
- @Override
- public String version() {
- return VERSION;
- }
-
- @Override
- public void start(Map<String, String> props) {
+ public SampleTransformation() { }
+ public SampleTransformation(R transformedRecord) {
+ this.transformedRecord = transformedRecord;
}
@Override
- public Class<? extends Task> taskClass() {
- return null;
+ public R apply(R record) {
+ return transformedRecord;
}
@Override
- public List<Map<String, String>> taskConfigs(int maxTasks) {
- return null;
+ public ConfigDef config() {
+ return new ConfigDef()
+ .define("subconfig", ConfigDef.Type.STRING, "default", ConfigDef.Importance.LOW, "docs");
}
@Override
- public void stop() {
-
+ public void close() {
+ closed = true;
}
@Override
- public ConfigDef config() {
- return new ConfigDef()
- .define("required", ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "required docs")
- .define("optional", ConfigDef.Type.STRING, "defaultVal", ConfigDef.Importance.HIGH, "optional docs");
- }
+ public void configure(Map<String, ?> configs) { }
+
}
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java
index 77f992e..65ab0c7 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java
@@ -1790,7 +1790,7 @@ public class WorkerSinkTaskTest {
@Test
public void testHeadersWithCustomConverter() throws Exception {
StringConverter stringConverter = new StringConverter();
- TestConverterWithHeaders testConverter = new TestConverterWithHeaders();
+ SampleConverterWithHeaders testConverter = new SampleConverterWithHeaders();
createTask(initialState, stringConverter, testConverter, stringConverter);
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
index e2e2366..78db83c 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
@@ -996,7 +996,7 @@ public class WorkerSourceTaskTest extends ThreadedTest {
@Test
public void testHeadersWithCustomConverter() throws Exception {
StringConverter stringConverter = new StringConverter();
- TestConverterWithHeaders testConverter = new TestConverterWithHeaders();
+ SampleConverterWithHeaders testConverter = new SampleConverterWithHeaders();
createWorkerTask(TargetState.STARTED, stringConverter, testConverter, stringConverter);
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/entities/PluginInfoTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/entities/PluginInfoTest.java
new file mode 100644
index 0000000..540c8a4
--- /dev/null
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/entities/PluginInfoTest.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.runtime.rest.entities;
+
+import org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader;
+import org.junit.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class PluginInfoTest {
+
+ @Test
+ public void testNoVersionFilter() {
+ PluginInfo.NoVersionFilter filter = new PluginInfo.NoVersionFilter();
+ assertFalse(filter.equals("1.0"));
+ assertFalse(filter.equals(new Object()));
+ assertFalse(filter.equals(null));
+ assertTrue(filter.equals(DelegatingClassLoader.UNDEFINED_VERSION));
+ }
+}
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResourceTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResourceTest.java
index 16ef798..59cf83c 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResourceTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResourceTest.java
@@ -16,9 +16,7 @@
*/
package org.apache.kafka.connect.runtime.rest.resources;
-import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
-import javax.ws.rs.core.HttpHeaders;
import org.apache.kafka.common.config.Config;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigDef.Importance;
@@ -28,91 +26,129 @@ import org.apache.kafka.common.config.ConfigDef.Width;
import org.apache.kafka.common.config.ConfigValue;
import org.apache.kafka.connect.connector.Connector;
import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.converters.LongConverter;
import org.apache.kafka.connect.runtime.AbstractHerder;
import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.Herder;
-import org.apache.kafka.connect.runtime.TestSinkConnector;
-import org.apache.kafka.connect.runtime.TestSourceConnector;
-import org.apache.kafka.connect.runtime.WorkerConfig;
+import org.apache.kafka.connect.runtime.SampleSinkConnector;
+import org.apache.kafka.connect.runtime.SampleSourceConnector;
+import org.apache.kafka.connect.runtime.distributed.DistributedHerder;
+import org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader;
import org.apache.kafka.connect.runtime.isolation.PluginClassLoader;
import org.apache.kafka.connect.runtime.isolation.PluginDesc;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
import org.apache.kafka.connect.runtime.isolation.Plugins;
-import org.apache.kafka.connect.runtime.rest.RestClient;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos;
import org.apache.kafka.connect.runtime.rest.entities.ConfigKeyInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConfigValueInfo;
-import org.apache.kafka.connect.runtime.rest.entities.ConnectorPluginInfo;
+import org.apache.kafka.connect.runtime.rest.entities.PluginInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorType;
-import org.apache.kafka.connect.sink.SinkConnector;
import org.apache.kafka.connect.source.SourceConnector;
-import org.apache.kafka.connect.tools.MockConnector;
+import org.apache.kafka.connect.storage.StringConverter;
import org.apache.kafka.connect.tools.MockSinkConnector;
import org.apache.kafka.connect.tools.MockSourceConnector;
import org.apache.kafka.connect.tools.SchemaSourceConnector;
import org.apache.kafka.connect.tools.VerifiableSinkConnector;
import org.apache.kafka.connect.tools.VerifiableSourceConnector;
+import org.apache.kafka.connect.transforms.RegexRouter;
+import org.apache.kafka.connect.transforms.TimestampConverter;
+import org.apache.kafka.connect.transforms.predicates.HasHeaderKey;
+import org.apache.kafka.connect.transforms.predicates.RecordIsTombstone;
import org.apache.kafka.connect.util.Callback;
-import org.easymock.Capture;
-import org.easymock.EasyMock;
-import org.easymock.IAnswer;
import org.junit.Before;
import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.powermock.api.easymock.PowerMock;
-import org.powermock.api.easymock.annotation.Mock;
-import org.powermock.core.classloader.annotations.PowerMockIgnore;
-import org.powermock.core.classloader.annotations.PrepareForTest;
-import org.powermock.modules.junit4.PowerMockRunner;
+import org.mockito.ArgumentCaptor;
import javax.ws.rs.BadRequestException;
import java.net.URL;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static java.util.Arrays.asList;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.anyBoolean;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
-@RunWith(PowerMockRunner.class)
-@PrepareForTest(RestClient.class)
-@PowerMockIgnore("javax.management.*")
public class ConnectorPluginsResourceTest {
- private static Map<String, String> props;
- private static Map<String, String> partialProps = new HashMap<>();
+ private static final Map<String, String> PROPS;
+ private static final Map<String, String> PARTIAL_PROPS = new HashMap<>();
static {
- partialProps.put("name", "test");
- partialProps.put("test.string.config", "testString");
- partialProps.put("test.int.config", "1");
- partialProps.put("test.list.config", "a,b");
-
- props = new HashMap<>(partialProps);
- props.put("connector.class", ConnectorPluginsResourceTestConnector.class.getSimpleName());
- props.put("plugin.path", "test.path");
+ PARTIAL_PROPS.put("name", "test");
+ PARTIAL_PROPS.put("test.string.config", "testString");
+ PARTIAL_PROPS.put("test.int.config", "1");
+ PARTIAL_PROPS.put("test.list.config", "a,b");
+
+ PROPS = new HashMap<>(PARTIAL_PROPS);
+ PROPS.put("connector.class", ConnectorPluginsResourceTestConnector.class.getSimpleName());
}
private static final ConfigInfos CONFIG_INFOS;
private static final ConfigInfos PARTIAL_CONFIG_INFOS;
private static final int ERROR_COUNT = 0;
private static final int PARTIAL_CONFIG_ERROR_COUNT = 1;
- private static final Set<PluginDesc<Connector>> CONNECTOR_PLUGINS = new TreeSet<>();
+ private static final Set<MockConnectorPluginDesc<?>> SINK_CONNECTOR_PLUGINS = new TreeSet<>();
+ private static final Set<MockConnectorPluginDesc<?>> SOURCE_CONNECTOR_PLUGINS = new TreeSet<>();
+ private static final Set<MockConnectorPluginDesc<?>> CONVERTER_PLUGINS = new TreeSet<>();
+ private static final Set<MockConnectorPluginDesc<?>> HEADER_CONVERTER_PLUGINS = new TreeSet<>();
+ private static final Set<MockConnectorPluginDesc<?>> TRANSFORMATION_PLUGINS = new TreeSet<>();
+ private static final Set<MockConnectorPluginDesc<?>> PREDICATE_PLUGINS = new TreeSet<>();
+
+ static {
+ try {
+ SINK_CONNECTOR_PLUGINS.add(new MockConnectorPluginDesc<>(VerifiableSinkConnector.class));
+ SINK_CONNECTOR_PLUGINS.add(new MockConnectorPluginDesc<>(MockSinkConnector.class));
+
+ SOURCE_CONNECTOR_PLUGINS.add(new MockConnectorPluginDesc<>(VerifiableSourceConnector.class));
+ SOURCE_CONNECTOR_PLUGINS.add(new MockConnectorPluginDesc<>(MockSourceConnector.class));
+ SOURCE_CONNECTOR_PLUGINS.add(new MockConnectorPluginDesc<>(SchemaSourceConnector.class));
+ SOURCE_CONNECTOR_PLUGINS.add(new MockConnectorPluginDesc<>(ConnectorPluginsResourceTestConnector.class));
+
+ CONVERTER_PLUGINS.add(new MockConnectorPluginDesc<>(StringConverter.class));
+ CONVERTER_PLUGINS.add(new MockConnectorPluginDesc<>(LongConverter.class));
+
+ HEADER_CONVERTER_PLUGINS.add(new MockConnectorPluginDesc<>(StringConverter.class));
+ HEADER_CONVERTER_PLUGINS.add(new MockConnectorPluginDesc<>(LongConverter.class));
+
+ TRANSFORMATION_PLUGINS.add(new MockConnectorPluginDesc<>(RegexRouter.class));
+ TRANSFORMATION_PLUGINS.add(new MockConnectorPluginDesc<>(TimestampConverter.class));
+
+ PREDICATE_PLUGINS.add(new MockConnectorPluginDesc<>(HasHeaderKey.class));
+ PREDICATE_PLUGINS.add(new MockConnectorPluginDesc<>(RecordIsTombstone.class));
+ } catch (Exception e) {
+ fail("Failed setting up plugins");
+ }
+ }
static {
List<ConfigInfo> configs = new LinkedList<>();
List<ConfigInfo> partialConfigs = new LinkedList<>();
ConfigDef connectorConfigDef = ConnectorConfig.configDef();
- List<ConfigValue> connectorConfigValues = connectorConfigDef.validate(props);
- List<ConfigValue> partialConnectorConfigValues = connectorConfigDef.validate(partialProps);
+ List<ConfigValue> connectorConfigValues = connectorConfigDef.validate(PROPS);
+ List<ConfigValue> partialConnectorConfigValues = connectorConfigDef.validate(PARTIAL_PROPS);
ConfigInfos result = AbstractHerder.generateResult(ConnectorPluginsResourceTestConnector.class.getName(), connectorConfigDef.configKeys(), connectorConfigValues, Collections.emptyList());
ConfigInfos partialResult = AbstractHerder.generateResult(ConnectorPluginsResourceTestConnector.class.getName(), connectorConfigDef.configKeys(), partialConnectorConfigValues, Collections.emptyList());
configs.addAll(result.values());
@@ -144,70 +180,34 @@ public class ConnectorPluginsResourceTest {
CONFIG_INFOS = new ConfigInfos(ConnectorPluginsResourceTestConnector.class.getName(), ERROR_COUNT, Collections.singletonList("Test"), configs);
PARTIAL_CONFIG_INFOS = new ConfigInfos(ConnectorPluginsResourceTestConnector.class.getName(), PARTIAL_CONFIG_ERROR_COUNT, Collections.singletonList("Test"), partialConfigs);
-
- List<Class<? extends Connector>> abstractConnectorClasses = asList(
- Connector.class,
- SourceConnector.class,
- SinkConnector.class
- );
-
- List<Class<? extends Connector>> connectorClasses = asList(
- VerifiableSourceConnector.class,
- VerifiableSinkConnector.class,
- MockSourceConnector.class,
- MockSinkConnector.class,
- MockConnector.class,
- SchemaSourceConnector.class,
- ConnectorPluginsResourceTestConnector.class
- );
-
- try {
- for (Class<? extends Connector> klass : abstractConnectorClasses) {
- MockConnectorPluginDesc pluginDesc = new MockConnectorPluginDesc(klass, "0.0.0");
- CONNECTOR_PLUGINS.add(pluginDesc);
- }
- for (Class<? extends Connector> klass : connectorClasses) {
- MockConnectorPluginDesc pluginDesc = new MockConnectorPluginDesc(klass);
- CONNECTOR_PLUGINS.add(pluginDesc);
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
}
- @Mock
- private Herder herder;
- @Mock
- private Plugins plugins;
+ private final Herder herder = mock(DistributedHerder.class);
+ private final Plugins plugins = mock(Plugins.class);
private ConnectorPluginsResource connectorPluginsResource;
@Before
public void setUp() throws Exception {
- PowerMock.mockStatic(RestClient.class,
- RestClient.class.getMethod("httpRequest", String.class, String.class, HttpHeaders.class, Object.class, TypeReference.class, WorkerConfig.class));
-
- plugins = PowerMock.createMock(Plugins.class);
- herder = PowerMock.createMock(AbstractHerder.class);
+ doReturn(plugins).when(herder).plugins();
+ doReturn(SINK_CONNECTOR_PLUGINS).when(plugins).sinkConnectors();
+ doReturn(SOURCE_CONNECTOR_PLUGINS).when(plugins).sourceConnectors();
+ doReturn(CONVERTER_PLUGINS).when(plugins).converters();
+ doReturn(HEADER_CONVERTER_PLUGINS).when(plugins).headerConverters();
+ doReturn(TRANSFORMATION_PLUGINS).when(plugins).transformations();
+ doReturn(PREDICATE_PLUGINS).when(plugins).predicates();
connectorPluginsResource = new ConnectorPluginsResource(herder);
}
- private void expectPlugins() {
- EasyMock.expect(herder.plugins()).andReturn(plugins);
- EasyMock.expect(plugins.connectors()).andReturn(CONNECTOR_PLUGINS);
- PowerMock.replayAll();
- }
-
@Test
public void testValidateConfigWithSingleErrorDueToMissingConnectorClassname() throws Throwable {
- Capture<Callback<ConfigInfos>> configInfosCallback = EasyMock.newCapture();
- herder.validateConnectorConfig(EasyMock.eq(partialProps), EasyMock.capture(configInfosCallback), EasyMock.anyBoolean());
-
- PowerMock.expectLastCall().andAnswer((IAnswer<Void>) () -> {
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<Callback<ConfigInfos>> configInfosCallback = ArgumentCaptor.forClass(Callback.class);
+ doAnswer(invocation -> {
ConfigDef connectorConfigDef = ConnectorConfig.configDef();
- List<ConfigValue> connectorConfigValues = connectorConfigDef.validate(partialProps);
+ List<ConfigValue> connectorConfigValues = connectorConfigDef.validate(PARTIAL_PROPS);
Connector connector = new ConnectorPluginsResourceTestConnector();
- Config config = connector.validate(partialProps);
+ Config config = connector.validate(PARTIAL_PROPS);
ConfigDef configDef = connector.config();
Map<String, ConfigDef.ConfigKey> configKeys = configDef.configKeys();
List<ConfigValue> configValues = config.configValues();
@@ -224,15 +224,13 @@ public class ConnectorPluginsResourceTest {
);
configInfosCallback.getValue().onCompletion(null, configInfos);
return null;
- });
-
- PowerMock.replayAll();
+ }).when(herder).validateConnectorConfig(eq(PARTIAL_PROPS), configInfosCallback.capture(), anyBoolean());
// This call to validateConfigs does not throw a BadRequestException because we've mocked
// validateConnectorConfig.
ConfigInfos configInfos = connectorPluginsResource.validateConfigs(
ConnectorPluginsResourceTestConnector.class.getSimpleName(),
- partialProps
+ PARTIAL_PROPS
);
assertEquals(PARTIAL_CONFIG_INFOS.name(), configInfos.name());
assertEquals(PARTIAL_CONFIG_INFOS.errorCount(), configInfos.errorCount());
@@ -241,21 +239,19 @@ public class ConnectorPluginsResourceTest {
new HashSet<>(PARTIAL_CONFIG_INFOS.values()),
new HashSet<>(configInfos.values())
);
-
- PowerMock.verifyAll();
+ verify(herder).validateConnectorConfig(eq(PARTIAL_PROPS), any(), anyBoolean());
}
@Test
public void testValidateConfigWithSimpleName() throws Throwable {
- Capture<Callback<ConfigInfos>> configInfosCallback = EasyMock.newCapture();
- herder.validateConnectorConfig(EasyMock.eq(props), EasyMock.capture(configInfosCallback), EasyMock.anyBoolean());
-
- PowerMock.expectLastCall().andAnswer((IAnswer<ConfigInfos>) () -> {
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<Callback<ConfigInfos>> configInfosCallback = ArgumentCaptor.forClass(Callback.class);
+ doAnswer(invocation -> {
ConfigDef connectorConfigDef = ConnectorConfig.configDef();
- List<ConfigValue> connectorConfigValues = connectorConfigDef.validate(props);
+ List<ConfigValue> connectorConfigValues = connectorConfigDef.validate(PROPS);
Connector connector = new ConnectorPluginsResourceTestConnector();
- Config config = connector.validate(props);
+ Config config = connector.validate(PROPS);
ConfigDef configDef = connector.config();
Map<String, ConfigDef.ConfigKey> configKeys = configDef.configKeys();
List<ConfigValue> configValues = config.configValues();
@@ -272,34 +268,30 @@ public class ConnectorPluginsResourceTest {
);
configInfosCallback.getValue().onCompletion(null, configInfos);
return null;
- });
-
- PowerMock.replayAll();
+ }).when(herder).validateConnectorConfig(eq(PROPS), configInfosCallback.capture(), anyBoolean());
// make a request to connector-plugins resource using just the simple class name.
ConfigInfos configInfos = connectorPluginsResource.validateConfigs(
ConnectorPluginsResourceTestConnector.class.getSimpleName(),
- props
+ PROPS
);
assertEquals(CONFIG_INFOS.name(), configInfos.name());
assertEquals(0, configInfos.errorCount());
assertEquals(CONFIG_INFOS.groups(), configInfos.groups());
assertEquals(new HashSet<>(CONFIG_INFOS.values()), new HashSet<>(configInfos.values()));
-
- PowerMock.verifyAll();
+ verify(herder).validateConnectorConfig(eq(PROPS), any(), anyBoolean());
}
@Test
public void testValidateConfigWithAlias() throws Throwable {
- Capture<Callback<ConfigInfos>> configInfosCallback = EasyMock.newCapture();
- herder.validateConnectorConfig(EasyMock.eq(props), EasyMock.capture(configInfosCallback), EasyMock.anyBoolean());
-
- PowerMock.expectLastCall().andAnswer((IAnswer<ConfigInfos>) () -> {
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<Callback<ConfigInfos>> configInfosCallback = ArgumentCaptor.forClass(Callback.class);
+ doAnswer(invocation -> {
ConfigDef connectorConfigDef = ConnectorConfig.configDef();
- List<ConfigValue> connectorConfigValues = connectorConfigDef.validate(props);
+ List<ConfigValue> connectorConfigValues = connectorConfigDef.validate(PROPS);
Connector connector = new ConnectorPluginsResourceTestConnector();
- Config config = connector.validate(props);
+ Config config = connector.validate(PROPS);
ConfigDef configDef = connector.config();
Map<String, ConfigDef.ConfigKey> configKeys = configDef.configKeys();
List<ConfigValue> configValues = config.configValues();
@@ -316,21 +308,18 @@ public class ConnectorPluginsResourceTest {
);
configInfosCallback.getValue().onCompletion(null, configInfos);
return null;
- });
-
- PowerMock.replayAll();
+ }).when(herder).validateConnectorConfig(eq(PROPS), configInfosCallback.capture(), anyBoolean());
// make a request to connector-plugins resource using a valid alias.
ConfigInfos configInfos = connectorPluginsResource.validateConfigs(
"ConnectorPluginsResourceTest",
- props
+ PROPS
);
assertEquals(CONFIG_INFOS.name(), configInfos.name());
assertEquals(0, configInfos.errorCount());
assertEquals(CONFIG_INFOS.groups(), configInfos.groups());
assertEquals(new HashSet<>(CONFIG_INFOS.values()), new HashSet<>(configInfos.values()));
-
- PowerMock.verifyAll();
+ verify(herder).validateConnectorConfig(eq(PROPS), any(), anyBoolean());
}
@Test
@@ -339,44 +328,39 @@ public class ConnectorPluginsResourceTest {
// simple name but different package.
String customClassname = "com.custom.package."
+ ConnectorPluginsResourceTestConnector.class.getSimpleName();
- assertThrows(BadRequestException.class, () -> connectorPluginsResource.validateConfigs(customClassname, props));
+ assertThrows(BadRequestException.class, () -> connectorPluginsResource.validateConfigs(customClassname, PROPS));
}
@Test
public void testValidateConfigWithNonExistentAlias() {
- assertThrows(BadRequestException.class, () -> connectorPluginsResource.validateConfigs("ConnectorPluginsTest", props));
+ assertThrows(BadRequestException.class, () -> connectorPluginsResource.validateConfigs("ConnectorPluginsTest", PROPS));
}
@Test
- public void testListConnectorPlugins() throws Exception {
- expectPlugins();
- Set<ConnectorPluginInfo> connectorPlugins = new HashSet<>(connectorPluginsResource.listConnectorPlugins());
- assertFalse(connectorPlugins.contains(newInfo(Connector.class, "0.0")));
- assertFalse(connectorPlugins.contains(newInfo(SourceConnector.class, "0.0")));
- assertFalse(connectorPlugins.contains(newInfo(SinkConnector.class, "0.0")));
- assertFalse(connectorPlugins.contains(newInfo(VerifiableSourceConnector.class)));
- assertFalse(connectorPlugins.contains(newInfo(VerifiableSinkConnector.class)));
- assertFalse(connectorPlugins.contains(newInfo(MockSourceConnector.class)));
- assertFalse(connectorPlugins.contains(newInfo(MockSinkConnector.class)));
- assertFalse(connectorPlugins.contains(newInfo(MockConnector.class)));
- assertFalse(connectorPlugins.contains(newInfo(SchemaSourceConnector.class)));
- assertTrue(connectorPlugins.contains(newInfo(ConnectorPluginsResourceTestConnector.class)));
- PowerMock.verifyAll();
+ public void testListConnectorPlugins() {
+ Set<Class<?>> excludes = Stream.of(ConnectorPluginsResource.SINK_CONNECTOR_EXCLUDES, ConnectorPluginsResource.SOURCE_CONNECTOR_EXCLUDES)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toSet());
+ Set<PluginInfo> expectedConnectorPlugins = Stream.of(SINK_CONNECTOR_PLUGINS, SOURCE_CONNECTOR_PLUGINS)
+ .flatMap(Collection::stream)
+ .filter(p -> !excludes.contains(p.pluginClass()))
+ .map(ConnectorPluginsResourceTest::newInfo)
+ .collect(Collectors.toSet());
+ Set<PluginInfo> actualConnectorPlugins = new HashSet<>(connectorPluginsResource.listConnectorPlugins(true));
+ assertEquals(expectedConnectorPlugins, actualConnectorPlugins);
+ verify(herder, atLeastOnce()).plugins();
}
@Test
- public void testConnectorPluginsIncludesTypeAndVersionInformation() throws Exception {
- expectPlugins();
- ConnectorPluginInfo sinkInfo = newInfo(TestSinkConnector.class);
- ConnectorPluginInfo sourceInfo =
- newInfo(TestSourceConnector.class);
- ConnectorPluginInfo unknownInfo =
- newInfo(ConnectorPluginsResourceTestConnector.class);
- assertEquals(ConnectorType.SINK, sinkInfo.type());
- assertEquals(ConnectorType.SOURCE, sourceInfo.type());
- assertEquals(ConnectorType.UNKNOWN, unknownInfo.type());
- assertEquals(TestSinkConnector.VERSION, sinkInfo.version());
- assertEquals(TestSourceConnector.VERSION, sourceInfo.version());
+ public void testConnectorPluginsIncludesClassTypeAndVersionInformation() throws Exception {
+ PluginInfo sinkInfo = newInfo(SampleSinkConnector.class);
+ PluginInfo sourceInfo = newInfo(SampleSourceConnector.class);
+ assertEquals(PluginType.SINK.toString(), sinkInfo.type());
+ assertEquals(PluginType.SOURCE.toString(), sourceInfo.type());
+ assertEquals(SampleSinkConnector.VERSION, sinkInfo.version());
+ assertEquals(SampleSourceConnector.VERSION, sourceInfo.version());
+ assertEquals(SampleSinkConnector.class.getName(), sinkInfo.className());
+ assertEquals(SampleSourceConnector.class.getName(), sourceInfo.className());
final ObjectMapper objectMapper = new ObjectMapper();
String serializedSink = objectMapper.writeValueAsString(ConnectorType.SINK);
@@ -399,19 +383,58 @@ public class ConnectorPluginsResourceTest {
);
}
- protected static ConnectorPluginInfo newInfo(Class<? extends Connector> klass, String version) {
- return new ConnectorPluginInfo(new MockConnectorPluginDesc(klass, version));
+ @Test
+ public void testListAllPlugins() {
+ Set<Class<?>> excludes = Stream.of(
+ ConnectorPluginsResource.SINK_CONNECTOR_EXCLUDES,
+ ConnectorPluginsResource.SOURCE_CONNECTOR_EXCLUDES,
+ ConnectorPluginsResource.TRANSFORM_EXCLUDES
+ ).flatMap(Collection::stream)
+ .collect(Collectors.toSet());
+ Set<PluginInfo> expectedConnectorPlugins = Stream.of(
+ SINK_CONNECTOR_PLUGINS,
+ SOURCE_CONNECTOR_PLUGINS,
+ CONVERTER_PLUGINS,
+ HEADER_CONVERTER_PLUGINS,
+ TRANSFORMATION_PLUGINS,
+ PREDICATE_PLUGINS
+ ).flatMap(Collection::stream)
+ .filter(p -> !excludes.contains(p.pluginClass()))
+ .map(ConnectorPluginsResourceTest::newInfo)
+ .collect(Collectors.toSet());
+ Set<PluginInfo> actualConnectorPlugins = new HashSet<>(connectorPluginsResource.listConnectorPlugins(false));
+ assertEquals(expectedConnectorPlugins, actualConnectorPlugins);
+ verify(herder, atLeastOnce()).plugins();
}
- protected static ConnectorPluginInfo newInfo(Class<? extends Connector> klass)
+ @Test
+ public void testGetConnectorConfigDef() {
+ String connName = ConnectorPluginsResourceTestConnector.class.getName();
+ when(herder.connectorPluginConfig(eq(connName))).thenAnswer(answer -> {
+ List<ConfigKeyInfo> results = new ArrayList<>();
+ for (ConfigDef.ConfigKey configKey : ConnectorPluginsResourceTestConnector.CONFIG_DEF.configKeys().values()) {
+ results.add(AbstractHerder.convertConfigKey(configKey));
+ }
+ return results;
+ });
+ List<ConfigKeyInfo> connectorConfigDef = connectorPluginsResource.getConnectorConfigDef(connName);
+ assertEquals(ConnectorPluginsResourceTestConnector.CONFIG_DEF.names().size(), connectorConfigDef.size());
+ for (String config : ConnectorPluginsResourceTestConnector.CONFIG_DEF.names()) {
+ Optional<ConfigKeyInfo> cki = connectorConfigDef.stream().filter(c -> c.name().equals(config)).findFirst();
+ assertTrue(cki.isPresent());
+ }
+ }
+
+ protected static PluginInfo newInfo(PluginDesc<?> pluginDesc) {
+ return new PluginInfo(new MockConnectorPluginDesc<>(pluginDesc.pluginClass(), pluginDesc.version()));
+ }
+
+ protected static PluginInfo newInfo(Class<?> klass)
throws Exception {
- return new ConnectorPluginInfo(new MockConnectorPluginDesc(klass));
+ return new PluginInfo(new MockConnectorPluginDesc<>(klass));
}
public static class MockPluginClassLoader extends PluginClassLoader {
- public MockPluginClassLoader(URL pluginLocation, URL[] urls, ClassLoader parent) {
- super(pluginLocation, urls, parent);
- }
public MockPluginClassLoader(URL pluginLocation, URL[] urls) {
super(pluginLocation, urls);
@@ -423,22 +446,22 @@ public class ConnectorPluginsResourceTest {
}
}
- public static class MockConnectorPluginDesc extends PluginDesc<Connector> {
- public MockConnectorPluginDesc(Class<? extends Connector> klass, String version) {
+ public static class MockConnectorPluginDesc<T> extends PluginDesc<T> {
+ public MockConnectorPluginDesc(Class<T> klass, String version) {
super(klass, version, new MockPluginClassLoader(null, new URL[0]));
}
- public MockConnectorPluginDesc(Class<? extends Connector> klass) throws Exception {
+ public MockConnectorPluginDesc(Class<T> klass) throws Exception {
super(
klass,
- klass.getConstructor().newInstance().version(),
+ DelegatingClassLoader.versionFor(klass),
new MockPluginClassLoader(null, new URL[0])
);
}
}
/* Name here needs to be unique as we are testing the aliasing mechanism */
- public static class ConnectorPluginsResourceTestConnector extends Connector {
+ public static class ConnectorPluginsResourceTestConnector extends SourceConnector {
private static final String TEST_STRING_CONFIG = "test.string.config";
private static final String TEST_INT_CONFIG = "test.int.config";
diff --git a/gradle/spotbugs-exclude.xml b/gradle/spotbugs-exclude.xml
index 9858811..8e09cf9 100644
--- a/gradle/spotbugs-exclude.xml
+++ b/gradle/spotbugs-exclude.xml
@@ -261,6 +261,14 @@ For a detailed description of spotbugs bug categories, see https://spotbugs.read
</Match>
<Match>
+ <!-- Suppress warning about missing reflexive and symmetric properties of equal.
+ This equals method is used as a value filter for Jackson. -->
+ <Class name="org.apache.kafka.connect.runtime.rest.entities.PluginInfo$NoVersionFilter"/>
+ <Method name="equals"/>
+ <Bug pattern="EQ_CHECK_FOR_OPERAND_NOT_COMPATIBLE_WITH_THIS"/>
+ </Match>
+
+ <Match>
<!-- Suppress some minor warnings about machine-generated code for benchmarking. -->
<Package name="~org\.apache\.kafka\.jmh\..*\.jmh_generated"/>
</Match>