You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gh...@apache.org on 2025/01/07 17:46:35 UTC
(kafka) branch 4.0 updated: KAFKA-18436: Revert Multiversioning Changes from 4.0 release. (#18411)
This is an automated email from the ASF dual-hosted git repository.
gharris pushed a commit to branch 4.0
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.0 by this push:
new e56b3ea2efa KAFKA-18436: Revert Multiversioning Changes from 4.0 release. (#18411)
e56b3ea2efa is described below
commit e56b3ea2efa99472d4f58a6a170c35ad42f737ce
Author: snehashisp <sn...@gmail.com>
AuthorDate: Tue Jan 7 23:16:28 2025 +0530
KAFKA-18436: Revert Multiversioning Changes from 4.0 release. (#18411)
These changes will be postponed to the 4.1 release.
* Revert "KAFKA-18215: KIP-891 Connect Multiversioning Support (Configs and Validation changes for Connectors and Converters) (#17741)"
This reverts commit f4fe6064cc2f52555dc5a39ded342ef439eb599a.
* Revert "KAFKA-18182: KIP-891 Add VersionRange to Plugins and DelegatingClassLoader APIs (#16984)"
This reverts commit af0054b50204c17be0cc84b8a3ef4575324bcb79.
Reviewers: Greg Harris <gr...@aiven.io>
---
checkstyle/import-control.xml | 2 +-
.../kafka/connect/runtime/AbstractHerder.java | 409 ++++++++-------------
.../kafka/connect/runtime/CachedConnectors.java | 79 ----
.../kafka/connect/runtime/ConnectorConfig.java | 238 ++----------
.../org/apache/kafka/connect/runtime/Herder.java | 4 -
.../kafka/connect/runtime/SinkConnectorConfig.java | 28 +-
.../connect/runtime/SourceConnectorConfig.java | 16 +-
.../org/apache/kafka/connect/runtime/Worker.java | 4 -
.../apache/kafka/connect/runtime/WorkerConfig.java | 20 -
.../runtime/isolation/DelegatingClassLoader.java | 184 +--------
.../connect/runtime/isolation/PluginDesc.java | 6 -
.../connect/runtime/isolation/PluginUtils.java | 23 +-
.../kafka/connect/runtime/isolation/Plugins.java | 312 +++++-----------
.../runtime/isolation/PluginsRecommenders.java | 198 ----------
.../isolation/VersionedPluginLoadingException.java | 40 --
.../rest/resources/ConnectorPluginsResource.java | 17 +-
.../kafka/connect/runtime/AbstractHerderTest.java | 66 ++--
.../connect/runtime/SampleSourceConnector.java | 2 +-
.../runtime/distributed/DistributedHerderTest.java | 6 +-
.../runtime/isolation/SynchronizationTest.java | 5 +-
.../resources/ConnectorPluginsResourceTest.java | 4 +-
.../runtime/standalone/StandaloneHerderTest.java | 30 +-
22 files changed, 373 insertions(+), 1320 deletions(-)
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index 88cbee167c2..280594902be 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -572,7 +572,6 @@
<allow pkg="org.apache.kafka.connect" />
<allow pkg="io.github.classgraph"/>
<allow pkg="javax.crypto"/>
- <allow pkg="org.apache.maven.artifact.versioning" />
<allow pkg="org.eclipse.jetty.util" />
<allow pkg="org.apache.logging.log4j" />
@@ -591,6 +590,7 @@
<subpackage name="isolation">
<allow pkg="com.fasterxml.jackson" />
+ <allow pkg="org.apache.maven.artifact.versioning" />
<allow pkg="javax.tools" />
</subpackage>
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 c4f53ad4137..046dcc63e5a 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
@@ -34,9 +34,7 @@ import org.apache.kafka.connect.connector.policy.ConnectorClientConfigRequest;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.errors.NotFoundException;
import org.apache.kafka.connect.runtime.isolation.LoaderSwap;
-import org.apache.kafka.connect.runtime.isolation.PluginUtils;
import org.apache.kafka.connect.runtime.isolation.Plugins;
-import org.apache.kafka.connect.runtime.isolation.VersionedPluginLoadingException;
import org.apache.kafka.connect.runtime.rest.entities.ActiveTopicsInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos;
@@ -66,8 +64,6 @@ import org.apache.kafka.connect.util.Stage;
import org.apache.kafka.connect.util.TemporaryStage;
import org.apache.logging.log4j.Level;
-import org.apache.maven.artifact.versioning.InvalidVersionSpecificationException;
-import org.apache.maven.artifact.versioning.VersionRange;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -89,6 +85,8 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Function;
@@ -96,15 +94,9 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
-import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLASS_CONFIG;
-import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_VERSION;
import static org.apache.kafka.connect.runtime.ConnectorConfig.HEADER_CONVERTER_CLASS_CONFIG;
-import static org.apache.kafka.connect.runtime.ConnectorConfig.HEADER_CONVERTER_VERSION_CONFIG;
import static org.apache.kafka.connect.runtime.ConnectorConfig.KEY_CONVERTER_CLASS_CONFIG;
-import static org.apache.kafka.connect.runtime.ConnectorConfig.KEY_CONVERTER_VERSION_CONFIG;
import static org.apache.kafka.connect.runtime.ConnectorConfig.VALUE_CONVERTER_CLASS_CONFIG;
-import static org.apache.kafka.connect.runtime.ConnectorConfig.VALUE_CONVERTER_VERSION_CONFIG;
-
/**
* Abstract Herder implementation which handles connector/task lifecycle tracking. Extensions
@@ -146,7 +138,7 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
private final Time time;
protected final Loggers loggers;
- private final CachedConnectors cachedConnectors;
+ private final ConcurrentMap<String, Connector> tempConnectors = new ConcurrentHashMap<>();
public AbstractHerder(Worker worker,
String workerId,
@@ -165,7 +157,6 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
this.connectorExecutor = Executors.newCachedThreadPool();
this.time = time;
this.loggers = new Loggers(time);
- this.cachedConnectors = new CachedConnectors(worker.getPlugins());
}
@Override
@@ -407,8 +398,6 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
* may be null, in which case no validation will be performed under the assumption that the
* connector will use inherit the converter settings from the worker. Some errors encountered
* during validation may be {@link ConfigValue#addErrorMessage(String) added} to this object
- * @param pluginVersionValue the {@link ConfigValue} for the converter version property in the connector config;
- *
* @param pluginInterface the interface for the plugin type
* (e.g., {@code org.apache.kafka.connect.storage.Converter.class});
* may not be null
@@ -429,18 +418,14 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
* @param <T> the plugin class to perform validation for
*/
- @SuppressWarnings("unchecked")
private <T> ConfigInfos validateConverterConfig(
Map<String, String> connectorConfig,
ConfigValue pluginConfigValue,
- ConfigValue pluginVersionValue,
Class<T> pluginInterface,
Function<T, ConfigDef> configDefAccessor,
String pluginName,
String pluginProperty,
- String pluginVersionProperty,
Map<String, String> defaultProperties,
- ClassLoader connectorLoader,
Function<String, TemporaryStage> reportStage
) {
Objects.requireNonNull(connectorConfig);
@@ -448,15 +433,12 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
Objects.requireNonNull(configDefAccessor);
Objects.requireNonNull(pluginName);
Objects.requireNonNull(pluginProperty);
- Objects.requireNonNull(pluginVersionProperty);
String pluginClass = connectorConfig.get(pluginProperty);
- String pluginVersion = connectorConfig.get(pluginVersionProperty);
if (pluginClass == null
|| pluginConfigValue == null
|| !pluginConfigValue.errorMessages().isEmpty()
- || !pluginVersionValue.errorMessages().isEmpty()
) {
// Either no custom converter was specified, or one was specified but there's a problem with it.
// No need to proceed any further.
@@ -466,22 +448,11 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
T pluginInstance;
String stageDescription = "instantiating the connector's " + pluginName + " for validation";
try (TemporaryStage stage = reportStage.apply(stageDescription)) {
- VersionRange range = PluginUtils.connectorVersionRequirement(pluginVersion);
- pluginInstance = (T) plugins().newPlugin(pluginClass, range, connectorLoader);
- } catch (VersionedPluginLoadingException e) {
- log.error("Failed to load {} class {} with version {}: {}", pluginName, pluginClass, pluginVersion, e);
- pluginConfigValue.addErrorMessage(e.getMessage());
- pluginVersionValue.addErrorMessage(e.getMessage());
- return null;
+ pluginInstance = Utils.newInstance(pluginClass, pluginInterface);
} catch (ClassNotFoundException | RuntimeException e) {
log.error("Failed to instantiate {} class {}; this should have been caught by prior validation logic", pluginName, pluginClass, e);
pluginConfigValue.addErrorMessage("Failed to load class " + pluginClass + (e.getMessage() != null ? ": " + e.getMessage() : ""));
return null;
- } catch (InvalidVersionSpecificationException e) {
- // this should have been caught by prior validation logic
- log.error("Invalid version range for {} class {}: {}", pluginName, pluginClass, pluginVersion, e);
- pluginVersionValue.addErrorMessage(e.getMessage());
- return null;
}
try {
@@ -523,55 +494,55 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
}
}
- private ConfigInfos validateAllConverterConfigs(
- Map<String, String> connectorProps,
- Map<String, ConfigValue> validatedConnectorConfig,
- ClassLoader connectorLoader,
+ private ConfigInfos validateHeaderConverterConfig(
+ Map<String, String> connectorConfig,
+ ConfigValue headerConverterConfigValue,
Function<String, TemporaryStage> reportStage
) {
- String connType = connectorProps.get(CONNECTOR_CLASS_CONFIG);
- // do custom converter-specific validation
- ConfigInfos headerConverterConfigInfos = validateConverterConfig(
- connectorProps,
- validatedConnectorConfig.get(HEADER_CONVERTER_CLASS_CONFIG),
- validatedConnectorConfig.get(HEADER_CONVERTER_VERSION_CONFIG),
+ return validateConverterConfig(
+ connectorConfig,
+ headerConverterConfigValue,
HeaderConverter.class,
HeaderConverter::config,
"header converter",
HEADER_CONVERTER_CLASS_CONFIG,
- HEADER_CONVERTER_VERSION_CONFIG,
Collections.singletonMap(ConverterConfig.TYPE_CONFIG, ConverterType.HEADER.getName()),
- connectorLoader,
reportStage
);
- ConfigInfos keyConverterConfigInfos = validateConverterConfig(
- connectorProps,
- validatedConnectorConfig.get(KEY_CONVERTER_CLASS_CONFIG),
- validatedConnectorConfig.get(KEY_CONVERTER_VERSION_CONFIG),
+ }
+
+ private ConfigInfos validateKeyConverterConfig(
+ Map<String, String> connectorConfig,
+ ConfigValue keyConverterConfigValue,
+ Function<String, TemporaryStage> reportStage
+ ) {
+ return validateConverterConfig(
+ connectorConfig,
+ keyConverterConfigValue,
Converter.class,
Converter::config,
"key converter",
KEY_CONVERTER_CLASS_CONFIG,
- KEY_CONVERTER_VERSION_CONFIG,
Collections.singletonMap(ConverterConfig.TYPE_CONFIG, ConverterType.KEY.getName()),
- connectorLoader,
reportStage
);
+ }
- ConfigInfos valueConverterConfigInfos = validateConverterConfig(
- connectorProps,
- validatedConnectorConfig.get(VALUE_CONVERTER_CLASS_CONFIG),
- validatedConnectorConfig.get(VALUE_CONVERTER_VERSION_CONFIG),
+ private ConfigInfos validateValueConverterConfig(
+ Map<String, String> connectorConfig,
+ ConfigValue valueConverterConfigValue,
+ Function<String, TemporaryStage> reportStage
+ ) {
+ return validateConverterConfig(
+ connectorConfig,
+ valueConverterConfigValue,
Converter.class,
Converter::config,
"value converter",
VALUE_CONVERTER_CLASS_CONFIG,
- VALUE_CONVERTER_VERSION_CONFIG,
Collections.singletonMap(ConverterConfig.TYPE_CONFIG, ConverterType.VALUE.getName()),
- connectorLoader,
reportStage
);
- return mergeConfigInfos(connType, headerConverterConfigInfos, keyConverterConfigInfos, valueConverterConfigInfos);
}
@Override
@@ -663,146 +634,6 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
|| SinkConnectorConfig.hasDlqTopicConfig(connProps);
}
- private ConfigInfos validateClientOverrides(
- Map<String, String> connectorProps,
- org.apache.kafka.connect.health.ConnectorType connectorType,
- Class<? extends Connector> connectorClass,
- Function<String, TemporaryStage> reportStage,
- boolean doLog
- ) {
- if (connectorClass == null || connectorType == null) {
- return null;
- }
- AbstractConfig connectorConfig = new AbstractConfig(new ConfigDef(), connectorProps, doLog);
- String connName = connectorProps.get(ConnectorConfig.NAME_CONFIG);
- String connType = connectorProps.get(CONNECTOR_CLASS_CONFIG);
- ConfigInfos producerConfigInfos = null;
- ConfigInfos consumerConfigInfos = null;
- ConfigInfos adminConfigInfos = null;
- String stageDescription = null;
-
- if (connectorUsesProducer(connectorType, connectorProps)) {
- stageDescription = "validating producer config overrides for the connector";
- try (TemporaryStage stage = reportStage.apply(stageDescription)) {
- producerConfigInfos = validateClientOverrides(
- connName,
- ConnectorConfig.CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX,
- connectorConfig,
- ProducerConfig.configDef(),
- connectorClass,
- connectorType,
- ConnectorClientConfigRequest.ClientType.PRODUCER,
- connectorClientConfigOverridePolicy);
- }
- }
- if (connectorUsesAdmin(connectorType, connectorProps)) {
- stageDescription = "validating admin config overrides for the connector";
- try (TemporaryStage stage = reportStage.apply(stageDescription)) {
- adminConfigInfos = validateClientOverrides(
- connName,
- ConnectorConfig.CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX,
- connectorConfig,
- AdminClientConfig.configDef(),
- connectorClass,
- connectorType,
- ConnectorClientConfigRequest.ClientType.ADMIN,
- connectorClientConfigOverridePolicy);
- }
- }
- if (connectorUsesConsumer(connectorType, connectorProps)) {
- stageDescription = "validating consumer config overrides for the connector";
- try (TemporaryStage stage = reportStage.apply(stageDescription)) {
- consumerConfigInfos = validateClientOverrides(
- connName,
- ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX,
- connectorConfig,
- ConsumerConfig.configDef(),
- connectorClass,
- connectorType,
- ConnectorClientConfigRequest.ClientType.CONSUMER,
- connectorClientConfigOverridePolicy);
- }
- }
- return mergeConfigInfos(connType,
- producerConfigInfos,
- consumerConfigInfos,
- adminConfigInfos
- );
- }
-
- private ConfigInfos validateConnectorPluginSpecifiedConfigs(
- Map<String, String> connectorProps,
- Map<String, ConfigValue> validatedConnectorConfig,
- ConfigDef enrichedConfigDef,
- Connector connector,
- Function<String, TemporaryStage> reportStage
- ) {
- List<ConfigValue> configValues = new ArrayList<>(validatedConnectorConfig.values());
- Map<String, ConfigKey> configKeys = new LinkedHashMap<>(enrichedConfigDef.configKeys());
- Set<String> allGroups = new LinkedHashSet<>(enrichedConfigDef.groups());
-
- String connType = connectorProps.get(CONNECTOR_CLASS_CONFIG);
- // do custom connector-specific validation
- ConfigDef configDef;
- String stageDescription = "retrieving the configuration definition from the connector";
- try (TemporaryStage stage = reportStage.apply(stageDescription)) {
- configDef = connector.config();
- }
- if (null == configDef) {
- throw new BadRequestException(
- String.format(
- "%s.config() must return a ConfigDef that is not null.",
- connector.getClass().getName()
- )
- );
- }
-
- Config config;
- stageDescription = "performing multi-property validation for the connector";
- try (TemporaryStage stage = reportStage.apply(stageDescription)) {
- config = connector.validate(connectorProps);
- }
- if (null == config) {
- throw new BadRequestException(
- String.format(
- "%s.validate() must return a Config that is not null.",
- connector.getClass().getName()
- )
- );
- }
- configKeys.putAll(configDef.configKeys());
- allGroups.addAll(configDef.groups());
- configValues.addAll(config.configValues());
- return generateResult(connType, configKeys, configValues, new ArrayList<>(allGroups));
- }
-
- private void addNullValuedErrors(Map<String, String> connectorProps, Map<String, ConfigValue> validatedConfig) {
- connectorProps.entrySet().stream()
- .filter(e -> e.getValue() == null)
- .map(Map.Entry::getKey)
- .forEach(prop ->
- validatedConfig.computeIfAbsent(prop, ConfigValue::new)
- .addErrorMessage("Null value can not be supplied as the configuration value."));
- }
-
- private ConfigInfos invalidVersionedConnectorValidation(
- Map<String, String> connectorProps,
- VersionedPluginLoadingException e,
- Function<String, TemporaryStage> reportStage
- ) {
- String connType = connectorProps.get(CONNECTOR_CLASS_CONFIG);
- ConfigDef configDef = ConnectorConfig.enrichedConfigDef(worker.getPlugins(), connType);
- Map<String, ConfigValue> validatedConfig;
- try (TemporaryStage stage = reportStage.apply("validating connector configuration")) {
- validatedConfig = configDef.validateAll(connectorProps);
- }
- validatedConfig.get(CONNECTOR_CLASS_CONFIG).addErrorMessage(e.getMessage());
- validatedConfig.get(CONNECTOR_VERSION).addErrorMessage(e.getMessage());
- validatedConfig.get(CONNECTOR_VERSION).recommendedValues(e.availableVersions().stream().map(v -> (Object) v).collect(Collectors.toList()));
- addNullValuedErrors(connectorProps, validatedConfig);
- return generateResult(connType, configDef.configKeys(), new ArrayList<>(validatedConfig.values()), new ArrayList<>(configDef.groups()));
- }
-
ConfigInfos validateConnectorConfig(
Map<String, String> connectorProps,
Function<String, TemporaryStage> reportStage,
@@ -815,58 +646,150 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
connectorProps = worker.configTransformer().transform(connectorProps);
}
}
- String connType = connectorProps.get(CONNECTOR_CLASS_CONFIG);
- if (connType == null) {
+ String connType = connectorProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
+ if (connType == null)
throw new BadRequestException("Connector config " + connectorProps + " contains no connector type");
- }
-
- VersionRange connVersion;
- Connector connector;
- ClassLoader connectorLoader;
- try {
- connVersion = PluginUtils.connectorVersionRequirement(connectorProps.get(CONNECTOR_VERSION));
- connector = cachedConnectors.getConnector(connType, connVersion);
- connectorLoader = plugins().pluginLoader(connType, connVersion);
- log.info("Validating connector {}, version {}", connType, connector.version());
- } catch (VersionedPluginLoadingException e) {
- log.warn("Failed to load connector {} with version {}, skipping additional validations (connector, converters, transformations, client overrides) ",
- connType, connectorProps.get(CONNECTOR_VERSION), e);
- return invalidVersionedConnectorValidation(connectorProps, e, reportStage);
- } catch (Exception e) {
- throw new BadRequestException(e.getMessage(), e);
- }
+ Connector connector = getConnector(connType);
+ ClassLoader connectorLoader = plugins().connectorLoader(connType);
try (LoaderSwap loaderSwap = plugins().withClassLoader(connectorLoader)) {
-
+ org.apache.kafka.connect.health.ConnectorType connectorType;
ConfigDef enrichedConfigDef;
Map<String, ConfigValue> validatedConnectorConfig;
- org.apache.kafka.connect.health.ConnectorType connectorType;
if (connector instanceof SourceConnector) {
connectorType = org.apache.kafka.connect.health.ConnectorType.SOURCE;
- enrichedConfigDef = ConnectorConfig.enrich(plugins(), SourceConnectorConfig.enrichedConfigDef(plugins(), connectorProps, worker.config()), connectorProps, false);
+ enrichedConfigDef = ConnectorConfig.enrich(plugins(), SourceConnectorConfig.configDef(), connectorProps, false);
stageDescription = "validating source connector-specific properties for the connector";
try (TemporaryStage stage = reportStage.apply(stageDescription)) {
validatedConnectorConfig = validateSourceConnectorConfig((SourceConnector) connector, enrichedConfigDef, connectorProps);
}
} else {
connectorType = org.apache.kafka.connect.health.ConnectorType.SINK;
- enrichedConfigDef = ConnectorConfig.enrich(plugins(), SinkConnectorConfig.enrichedConfigDef(plugins(), connectorProps, worker.config()), connectorProps, false);
+ enrichedConfigDef = ConnectorConfig.enrich(plugins(), SinkConnectorConfig.configDef(), connectorProps, false);
stageDescription = "validating sink connector-specific properties for the connector";
try (TemporaryStage stage = reportStage.apply(stageDescription)) {
validatedConnectorConfig = validateSinkConnectorConfig((SinkConnector) connector, enrichedConfigDef, connectorProps);
}
}
- addNullValuedErrors(connectorProps, validatedConnectorConfig);
+ connectorProps.entrySet().stream()
+ .filter(e -> e.getValue() == null)
+ .map(Map.Entry::getKey)
+ .forEach(prop ->
+ validatedConnectorConfig.computeIfAbsent(prop, ConfigValue::new)
+ .addErrorMessage("Null value can not be supplied as the configuration value.")
+ );
+
+ List<ConfigValue> configValues = new ArrayList<>(validatedConnectorConfig.values());
+ Map<String, ConfigKey> configKeys = new LinkedHashMap<>(enrichedConfigDef.configKeys());
+ Set<String> allGroups = new LinkedHashSet<>(enrichedConfigDef.groups());
- ConfigInfos connectorConfigInfo = validateConnectorPluginSpecifiedConfigs(connectorProps, validatedConnectorConfig, enrichedConfigDef, connector, reportStage);
- ConfigInfos converterConfigInfo = validateAllConverterConfigs(connectorProps, validatedConnectorConfig, connectorLoader, reportStage);
- ConfigInfos clientOverrideInfo = validateClientOverrides(connectorProps, connectorType, connector.getClass(), reportStage, doLog);
+ // do custom connector-specific validation
+ ConfigDef configDef;
+ stageDescription = "retrieving the configuration definition from the connector";
+ try (TemporaryStage stage = reportStage.apply(stageDescription)) {
+ configDef = connector.config();
+ }
+ if (null == configDef) {
+ throw new BadRequestException(
+ String.format(
+ "%s.config() must return a ConfigDef that is not null.",
+ connector.getClass().getName()
+ )
+ );
+ }
+ Config config;
+ stageDescription = "performing multi-property validation for the connector";
+ try (TemporaryStage stage = reportStage.apply(stageDescription)) {
+ config = connector.validate(connectorProps);
+ }
+ if (null == config) {
+ throw new BadRequestException(
+ String.format(
+ "%s.validate() must return a Config that is not null.",
+ connector.getClass().getName()
+ )
+ );
+ }
+ configKeys.putAll(configDef.configKeys());
+ allGroups.addAll(configDef.groups());
+ configValues.addAll(config.configValues());
+
+ // do custom converter-specific validation
+ ConfigInfos headerConverterConfigInfos = validateHeaderConverterConfig(
+ connectorProps,
+ validatedConnectorConfig.get(HEADER_CONVERTER_CLASS_CONFIG),
+ reportStage
+ );
+ ConfigInfos keyConverterConfigInfos = validateKeyConverterConfig(
+ connectorProps,
+ validatedConnectorConfig.get(KEY_CONVERTER_CLASS_CONFIG),
+ reportStage
+ );
+ ConfigInfos valueConverterConfigInfos = validateValueConverterConfig(
+ connectorProps,
+ validatedConnectorConfig.get(VALUE_CONVERTER_CLASS_CONFIG),
+ reportStage
+ );
+
+ ConfigInfos configInfos = generateResult(connType, configKeys, configValues, new ArrayList<>(allGroups));
+ AbstractConfig connectorConfig = new AbstractConfig(new ConfigDef(), connectorProps, doLog);
+ String connName = connectorProps.get(ConnectorConfig.NAME_CONFIG);
+ ConfigInfos producerConfigInfos = null;
+ ConfigInfos consumerConfigInfos = null;
+ ConfigInfos adminConfigInfos = null;
+
+ if (connectorUsesProducer(connectorType, connectorProps)) {
+ stageDescription = "validating producer config overrides for the connector";
+ try (TemporaryStage stage = reportStage.apply(stageDescription)) {
+ producerConfigInfos = validateClientOverrides(
+ connName,
+ ConnectorConfig.CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX,
+ connectorConfig,
+ ProducerConfig.configDef(),
+ connector.getClass(),
+ connectorType,
+ ConnectorClientConfigRequest.ClientType.PRODUCER,
+ connectorClientConfigOverridePolicy);
+ }
+ }
+ if (connectorUsesAdmin(connectorType, connectorProps)) {
+ stageDescription = "validating admin config overrides for the connector";
+ try (TemporaryStage stage = reportStage.apply(stageDescription)) {
+ adminConfigInfos = validateClientOverrides(
+ connName,
+ ConnectorConfig.CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX,
+ connectorConfig,
+ AdminClientConfig.configDef(),
+ connector.getClass(),
+ connectorType,
+ ConnectorClientConfigRequest.ClientType.ADMIN,
+ connectorClientConfigOverridePolicy);
+ }
+ }
+ if (connectorUsesConsumer(connectorType, connectorProps)) {
+ stageDescription = "validating consumer config overrides for the connector";
+ try (TemporaryStage stage = reportStage.apply(stageDescription)) {
+ consumerConfigInfos = validateClientOverrides(
+ connName,
+ ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX,
+ connectorConfig,
+ ConsumerConfig.configDef(),
+ connector.getClass(),
+ connectorType,
+ ConnectorClientConfigRequest.ClientType.CONSUMER,
+ connectorClientConfigOverridePolicy);
+ }
+ }
return mergeConfigInfos(connType,
- connectorConfigInfo,
- clientOverrideInfo,
- converterConfigInfo
+ configInfos,
+ producerConfigInfos,
+ consumerConfigInfos,
+ adminConfigInfos,
+ headerConverterConfigInfos,
+ keyConverterConfigInfos,
+ valueConverterConfigInfos
);
}
}
@@ -1013,6 +936,10 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
return new ConfigValueInfo(configValue.name(), value, recommendedValues, configValue.errorMessages(), configValue.visible());
}
+ protected Connector getConnector(String connType) {
+ return tempConnectors.computeIfAbsent(connType, k -> plugins().newConnector(k));
+ }
+
/**
* Retrieves ConnectorType for the class specified in the connector config
* @param connConfig the connector config, may be null
@@ -1023,14 +950,13 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
if (connConfig == null) {
return ConnectorType.UNKNOWN;
}
- String connClass = connConfig.get(CONNECTOR_CLASS_CONFIG);
+ String connClass = connConfig.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
if (connClass == null) {
return ConnectorType.UNKNOWN;
}
try {
- VersionRange range = PluginUtils.connectorVersionRequirement(connConfig.get(CONNECTOR_VERSION));
- return ConnectorType.from(cachedConnectors.getConnector(connClass, range).getClass());
- } catch (Exception e) {
+ return ConnectorType.from(getConnector(connClass).getClass());
+ } catch (ConnectException e) {
log.warn("Unable to retrieve connector type", e);
return ConnectorType.UNKNOWN;
}
@@ -1152,33 +1078,25 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
@Override
public List<ConfigKeyInfo> connectorPluginConfig(String pluginName) {
- return connectorPluginConfig(pluginName, null);
- }
-
- @Override
- public List<ConfigKeyInfo> connectorPluginConfig(String pluginName, VersionRange range) {
-
Plugins p = plugins();
Class<?> pluginClass;
try {
- pluginClass = p.pluginClass(pluginName, range);
+ pluginClass = p.pluginClass(pluginName);
} catch (ClassNotFoundException cnfe) {
throw new NotFoundException("Unknown plugin " + pluginName + ".");
- } catch (VersionedPluginLoadingException e) {
- throw new BadRequestException(e.getMessage(), e);
}
try (LoaderSwap loaderSwap = p.withClassLoader(pluginClass.getClassLoader())) {
- Object plugin = p.newPlugin(pluginName, range);
+ Object plugin = p.newPlugin(pluginName);
// Contains definitions coming from Connect framework
ConfigDef baseConfigDefs = null;
// Contains definitions specifically declared on the plugin
ConfigDef pluginConfigDefs;
if (plugin instanceof SinkConnector) {
- baseConfigDefs = SinkConnectorConfig.enrichedConfigDef(p, pluginName);
+ baseConfigDefs = SinkConnectorConfig.configDef();
pluginConfigDefs = ((SinkConnector) plugin).config();
} else if (plugin instanceof SourceConnector) {
- baseConfigDefs = SourceConnectorConfig.enrichedConfigDef(p, pluginName);
+ baseConfigDefs = SourceConnectorConfig.configDef();
pluginConfigDefs = ((SourceConnector) plugin).config();
} else if (plugin instanceof Converter) {
pluginConfigDefs = ((Converter) plugin).config();
@@ -1196,9 +1114,8 @@ public abstract class AbstractHerder implements Herder, TaskStatus.Listener, Con
// give precedence to the one defined by the plugin class
// Preserve the ordering of properties as they're returned from each ConfigDef
Map<String, ConfigKey> configsMap = new LinkedHashMap<>(pluginConfigDefs.configKeys());
- if (baseConfigDefs != null) {
+ if (baseConfigDefs != null)
baseConfigDefs.configKeys().forEach(configsMap::putIfAbsent);
- }
List<ConfigKeyInfo> results = new ArrayList<>();
for (ConfigKey configKey : configsMap.values()) {
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/CachedConnectors.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/CachedConnectors.java
deleted file mode 100644
index 59c4281fff8..00000000000
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/CachedConnectors.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * 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;
-
-import org.apache.kafka.connect.connector.Connector;
-import org.apache.kafka.connect.runtime.isolation.Plugins;
-import org.apache.kafka.connect.runtime.isolation.VersionedPluginLoadingException;
-
-import org.apache.maven.artifact.versioning.VersionRange;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-public class CachedConnectors {
-
- private static final String LATEST_VERSION = "latest";
-
- private final Map<String, Map<String, Connector>> connectors;
- private final Map<String, Exception> invalidConnectors;
- private final Map<String, Map<String, Exception>> invalidVersions;
- private final Plugins plugins;
-
- public CachedConnectors(Plugins plugins) {
- this.plugins = plugins;
- this.connectors = new ConcurrentHashMap<>();
- this.invalidConnectors = new ConcurrentHashMap<>();
- this.invalidVersions = new ConcurrentHashMap<>();
- }
-
- private void validate(String connectorName, VersionRange range) throws Exception {
- if (invalidConnectors.containsKey(connectorName)) {
- throw new Exception(invalidConnectors.get(connectorName));
- }
-
- String version = range == null ? LATEST_VERSION : range.toString();
- if (invalidVersions.containsKey(connectorName) && invalidVersions.get(connectorName).containsKey(version)) {
- throw new Exception(invalidVersions.get(connectorName).get(version));
- }
- }
-
- private Connector lookup(String connectorName, VersionRange range) throws Exception {
- String version = range == null ? LATEST_VERSION : range.toString();
- if (connectors.containsKey(connectorName) && connectors.get(connectorName).containsKey(version)) {
- return connectors.get(connectorName).get(version);
- }
-
- try {
- Connector connector = plugins.newConnector(connectorName, range);
- connectors.computeIfAbsent(connectorName, k -> new ConcurrentHashMap<>()).put(version, connector);
- return connector;
- } catch (VersionedPluginLoadingException e) {
- invalidVersions.computeIfAbsent(connectorName, k -> new ConcurrentHashMap<>()).put(version, e);
- throw e;
- } catch (Exception e) {
- invalidConnectors.put(connectorName, e);
- throw e;
- }
- }
-
- public Connector getConnector(String connectorName, VersionRange range) throws Exception {
- validate(connectorName, range);
- return lookup(connectorName, range);
- }
-}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java
index 929dc57a377..cb604ad73ee 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java
@@ -27,10 +27,7 @@ import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.runtime.errors.ToleranceType;
import org.apache.kafka.connect.runtime.isolation.PluginDesc;
-import org.apache.kafka.connect.runtime.isolation.PluginUtils;
import org.apache.kafka.connect.runtime.isolation.Plugins;
-import org.apache.kafka.connect.runtime.isolation.PluginsRecommenders;
-import org.apache.kafka.connect.runtime.isolation.VersionedPluginLoadingException;
import org.apache.kafka.connect.storage.Converter;
import org.apache.kafka.connect.storage.HeaderConverter;
import org.apache.kafka.connect.transforms.Transformation;
@@ -38,8 +35,6 @@ import org.apache.kafka.connect.transforms.predicates.Predicate;
import org.apache.kafka.connect.util.ConcreteSubClassValidator;
import org.apache.kafka.connect.util.InstantiableClassValidator;
-import org.apache.maven.artifact.versioning.InvalidVersionSpecificationException;
-import org.apache.maven.artifact.versioning.VersionRange;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -87,11 +82,6 @@ public class ConnectorConfig extends AbstractConfig {
" or use \"FileStreamSink\" or \"FileStreamSinkConnector\" to make the configuration a bit shorter";
private static final String CONNECTOR_CLASS_DISPLAY = "Connector class";
- public static final String CONNECTOR_VERSION = "connector." + WorkerConfig.PLUGIN_VERSION_SUFFIX;
- private static final String CONNECTOR_VERSION_DOC = "Version of the connector.";
- private static final String CONNECTOR_VERSION_DISPLAY = "Connector version";
- private static final ConfigDef.Validator CONNECTOR_VERSION_VALIDATOR = new PluginVersionValidator();
-
public static final String KEY_CONVERTER_CLASS_CONFIG = WorkerConfig.KEY_CONVERTER_CLASS_CONFIG;
public static final String KEY_CONVERTER_CLASS_DOC = WorkerConfig.KEY_CONVERTER_CLASS_DOC;
public static final String KEY_CONVERTER_CLASS_DISPLAY = "Key converter class";
@@ -100,12 +90,6 @@ public class ConnectorConfig extends AbstractConfig {
new InstantiableClassValidator()
);
- public static final String KEY_CONVERTER_VERSION_CONFIG = WorkerConfig.KEY_CONVERTER_VERSION;
- private static final String KEY_CONVERTER_VERSION_DOC = "Version of the key converter.";
- private static final String KEY_CONVERTER_VERSION_DISPLAY = "Key converter version";
- private static final ConfigDef.Validator KEY_CONVERTER_VERSION_VALIDATOR = new PluginVersionValidator();
-
-
public static final String VALUE_CONVERTER_CLASS_CONFIG = WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG;
public static final String VALUE_CONVERTER_CLASS_DOC = WorkerConfig.VALUE_CONVERTER_CLASS_DOC;
public static final String VALUE_CONVERTER_CLASS_DISPLAY = "Value converter class";
@@ -114,24 +98,17 @@ public class ConnectorConfig extends AbstractConfig {
new InstantiableClassValidator()
);
- public static final String VALUE_CONVERTER_VERSION_CONFIG = WorkerConfig.VALUE_CONVERTER_VERSION;
- private static final String VALUE_CONVERTER_VERSION_DOC = "Version of the value converter.";
- private static final String VALUE_CONVERTER_VERSION_DISPLAY = "Value converter version";
- private static final ConfigDef.Validator VALUE_CONVERTER_VERSION_VALIDATOR = new PluginVersionValidator();
-
public static final String HEADER_CONVERTER_CLASS_CONFIG = WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG;
public static final String HEADER_CONVERTER_CLASS_DOC = WorkerConfig.HEADER_CONVERTER_CLASS_DOC;
public static final String HEADER_CONVERTER_CLASS_DISPLAY = "Header converter class";
+ // The Connector config should not have a default for the header converter, since the absence of a config property means that
+ // the worker config settings should be used. Thus, we set the default to null here.
+ public static final String HEADER_CONVERTER_CLASS_DEFAULT = null;
private static final ConfigDef.Validator HEADER_CONVERTER_CLASS_VALIDATOR = ConfigDef.CompositeValidator.of(
ConcreteSubClassValidator.forSuperClass(HeaderConverter.class),
new InstantiableClassValidator()
);
- public static final String HEADER_CONVERTER_VERSION_CONFIG = WorkerConfig.HEADER_CONVERTER_VERSION;
- private static final String HEADER_CONVERTER_VERSION_DOC = "Version of the header converter.";
- private static final String HEADER_CONVERTER_VERSION_DISPLAY = "Header converter version";
- private static final ConfigDef.Validator HEADER_CONVERTER_VERSION_VALIDATOR = new PluginVersionValidator();
-
public static final String TASKS_MAX_CONFIG = "tasks.max";
private static final String TASKS_MAX_DOC = "Maximum number of tasks to use for this connector.";
public static final int TASKS_MAX_DEFAULT = 1;
@@ -210,11 +187,7 @@ public class ConnectorConfig extends AbstractConfig {
public static final String CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX = "admin.override.";
public static final String PREDICATES_PREFIX = "predicates.";
- private static final PluginsRecommenders EMPTY_RECOMMENDER = new PluginsRecommenders();
- private static final ConverterDefaults CONVERTER_DEFAULTS = new ConverterDefaults(null, null);
-
- private final ConnectorConfig.EnrichedConnectorConfig enrichedConfig;
-
+ private final EnrichedConnectorConfig enrichedConfig;
private static class EnrichedConnectorConfig extends AbstractConfig {
EnrichedConnectorConfig(ConfigDef configDef, Map<String, String> props) {
super(configDef, props);
@@ -226,27 +199,17 @@ public class ConnectorConfig extends AbstractConfig {
}
}
- protected static ConfigDef configDef(
- String defaultConnectorVersion,
- ConverterDefaults keyConverterDefaults,
- ConverterDefaults valueConverterDefaults,
- ConverterDefaults headerConverterDefaults,
- PluginsRecommenders recommender
- ) {
+ public static ConfigDef configDef() {
int orderInGroup = 0;
int orderInErrorGroup = 0;
return new ConfigDef()
.define(NAME_CONFIG, Type.STRING, ConfigDef.NO_DEFAULT_VALUE, nonEmptyStringWithoutControlChars(), Importance.HIGH, NAME_DOC, COMMON_GROUP, ++orderInGroup, Width.MEDIUM, NAME_DISPLAY)
.define(CONNECTOR_CLASS_CONFIG, Type.STRING, Importance.HIGH, CONNECTOR_CLASS_DOC, COMMON_GROUP, ++orderInGroup, Width.LONG, CONNECTOR_CLASS_DISPLAY)
- .define(CONNECTOR_VERSION, Type.STRING, defaultConnectorVersion, CONNECTOR_VERSION_VALIDATOR, Importance.MEDIUM, CONNECTOR_VERSION_DOC, COMMON_GROUP, ++orderInGroup, Width.MEDIUM, CONNECTOR_VERSION_DISPLAY, recommender.connectorPluginVersionRecommender())
.define(TASKS_MAX_CONFIG, Type.INT, TASKS_MAX_DEFAULT, atLeast(TASKS_MIN_CONFIG), Importance.HIGH, TASKS_MAX_DOC, COMMON_GROUP, ++orderInGroup, Width.SHORT, TASK_MAX_DISPLAY)
.define(TASKS_MAX_ENFORCE_CONFIG, Type.BOOLEAN, TASKS_MAX_ENFORCE_DEFAULT, Importance.LOW, TASKS_MAX_ENFORCE_DOC, COMMON_GROUP, ++orderInGroup, Width.SHORT, TASKS_MAX_ENFORCE_DISPLAY)
- .define(KEY_CONVERTER_CLASS_CONFIG, Type.CLASS, keyConverterDefaults.type, KEY_CONVERTER_CLASS_VALIDATOR, Importance.LOW, KEY_CONVERTER_CLASS_DOC, COMMON_GROUP, ++orderInGroup, Width.SHORT, KEY_CONVERTER_CLASS_DISPLAY, recommender.converterPluginRecommender())
- .define(KEY_CONVERTER_VERSION_CONFIG, Type.STRING, keyConverterDefaults.version, KEY_CONVERTER_VERSION_VALIDATOR, Importance.LOW, KEY_CONVERTER_VERSION_DOC, COMMON_GROUP, ++orderInGroup, Width.SHORT, KEY_CONVERTER_VERSION_DISPLAY, recommender.keyConverterPluginVersionRecommender())
- .define(VALUE_CONVERTER_CLASS_CONFIG, Type.CLASS, valueConverterDefaults.type, VALUE_CONVERTER_CLASS_VALIDATOR, Importance.LOW, VALUE_CONVERTER_CLASS_DOC, COMMON_GROUP, ++orderInGroup, Width.SHORT, VALUE_CONVERTER_CLASS_DISPLAY, recommender.converterPluginRecommender())
- .define(VALUE_CONVERTER_VERSION_CONFIG, Type.STRING, valueConverterDefaults.version, VALUE_CONVERTER_VERSION_VALIDATOR, Importance.LOW, VALUE_CONVERTER_VERSION_DOC, COMMON_GROUP, ++orderInGroup, Width.SHORT, VALUE_CONVERTER_VERSION_DISPLAY, recommender.valueConverterPluginVersionRecommender())
- .define(HEADER_CONVERTER_CLASS_CONFIG, Type.CLASS, headerConverterDefaults.type, HEADER_CONVERTER_CLASS_VALIDATOR, Importance.LOW, HEADER_CONVERTER_CLASS_DOC, COMMON_GROUP, ++orderInGroup, Width.SHORT, HEADER_CONVERTER_CLASS_DISPLAY, recommender.headerConverterPluginRecommender())
- .define(HEADER_CONVERTER_VERSION_CONFIG, Type.STRING, headerConverterDefaults.version, HEADER_CONVERTER_VERSION_VALIDATOR, Importance.LOW, HEADER_CONVERTER_VERSION_DOC, COMMON_GROUP, ++orderInGroup, Width.SHORT, HEADER_CONVERTER_VERSION_DISPLAY, recommender.headerConverterPluginVersionRecommender())
+ .define(KEY_CONVERTER_CLASS_CONFIG, Type.CLASS, null, KEY_CONVERTER_CLASS_VALIDATOR, Importance.LOW, KEY_CONVERTER_CLASS_DOC, COMMON_GROUP, ++orderInGroup, Width.SHORT, KEY_CONVERTER_CLASS_DISPLAY)
+ .define(VALUE_CONVERTER_CLASS_CONFIG, Type.CLASS, null, VALUE_CONVERTER_CLASS_VALIDATOR, Importance.LOW, VALUE_CONVERTER_CLASS_DOC, COMMON_GROUP, ++orderInGroup, Width.SHORT, VALUE_CONVERTER_CLASS_DISPLAY)
+ .define(HEADER_CONVERTER_CLASS_CONFIG, Type.CLASS, HEADER_CONVERTER_CLASS_DEFAULT, HEADER_CONVERTER_CLASS_VALIDATOR, Importance.LOW, HEADER_CONVERTER_CLASS_DOC, COMMON_GROUP, ++orderInGroup, Width.SHORT, HEADER_CONVERTER_CLASS_DISPLAY)
.define(TRANSFORMS_CONFIG, Type.LIST, Collections.emptyList(), aliasValidator("transformation"), Importance.LOW, TRANSFORMS_DOC, TRANSFORMS_GROUP, ++orderInGroup, Width.LONG, TRANSFORMS_DISPLAY)
.define(PREDICATES_CONFIG, Type.LIST, Collections.emptyList(), aliasValidator("predicate"), Importance.LOW, PREDICATES_DOC, PREDICATES_GROUP, ++orderInGroup, Width.LONG, PREDICATES_DISPLAY)
.define(CONFIG_RELOAD_ACTION_CONFIG, Type.STRING, CONFIG_RELOAD_ACTION_RESTART,
@@ -263,28 +226,6 @@ public class ConnectorConfig extends AbstractConfig {
ERRORS_LOG_ENABLE_DOC, ERROR_GROUP, ++orderInErrorGroup, Width.SHORT, ERRORS_LOG_ENABLE_DISPLAY)
.define(ERRORS_LOG_INCLUDE_MESSAGES_CONFIG, Type.BOOLEAN, ERRORS_LOG_INCLUDE_MESSAGES_DEFAULT, Importance.MEDIUM,
ERRORS_LOG_INCLUDE_MESSAGES_DOC, ERROR_GROUP, ++orderInErrorGroup, Width.SHORT, ERRORS_LOG_INCLUDE_MESSAGES_DISPLAY);
-
- }
-
- public static ConfigDef configDef() {
- return configDef(null, CONVERTER_DEFAULTS, CONVERTER_DEFAULTS, CONVERTER_DEFAULTS, EMPTY_RECOMMENDER);
- }
-
- // ConfigDef with additional defaults and recommenders
- public static ConfigDef enrichedConfigDef(Plugins plugins, Map<String, String> connProps, WorkerConfig workerConfig) {
- PluginsRecommenders recommender = new PluginsRecommenders(plugins);
- ConverterDefaults keyConverterDefaults = converterDefaults(plugins, KEY_CONVERTER_CLASS_CONFIG,
- WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, WorkerConfig.KEY_CONVERTER_VERSION, connProps, workerConfig, Converter.class);
- ConverterDefaults valueConverterDefaults = converterDefaults(plugins, VALUE_CONVERTER_CLASS_CONFIG,
- WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, WorkerConfig.VALUE_CONVERTER_VERSION, connProps, workerConfig, Converter.class);
- ConverterDefaults headerConverterDefaults = converterDefaults(plugins, HEADER_CONVERTER_CLASS_CONFIG,
- WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG, WorkerConfig.HEADER_CONVERTER_VERSION, connProps, workerConfig, HeaderConverter.class);
- return configDef(plugins.latestVersion(connProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG)),
- keyConverterDefaults, valueConverterDefaults, headerConverterDefaults, recommender);
- }
-
- public static ConfigDef enrichedConfigDef(Plugins plugins, String connectorClass) {
- return configDef(plugins.latestVersion(connectorClass), CONVERTER_DEFAULTS, CONVERTER_DEFAULTS, CONVERTER_DEFAULTS, EMPTY_RECOMMENDER);
}
private static ConfigDef.CompositeValidator aliasValidator(String kind) {
@@ -330,7 +271,7 @@ public class ConnectorConfig extends AbstractConfig {
public ToleranceType errorToleranceType() {
String tolerance = getString(ERRORS_TOLERANCE_CONFIG);
- for (ToleranceType type : ToleranceType.values()) {
+ for (ToleranceType type: ToleranceType.values()) {
if (type.name().equalsIgnoreCase(tolerance)) {
return type;
}
@@ -419,17 +360,17 @@ public class ConnectorConfig extends AbstractConfig {
@Override
protected Stream<Map.Entry<String, ConfigDef.ConfigKey>> configDefsForClass(String typeConfig) {
return super.configDefsForClass(typeConfig)
- .filter(entry -> {
- // The implicit parameters mask any from the transformer with the same name
- if (TransformationStage.PREDICATE_CONFIG.equals(entry.getKey())
- || TransformationStage.NEGATE_CONFIG.equals(entry.getKey())) {
- log.warn("Transformer config {} is masked by implicit config of that name",
- entry.getKey());
- return false;
- } else {
- return true;
- }
- });
+ .filter(entry -> {
+ // The implicit parameters mask any from the transformer with the same name
+ if (TransformationStage.PREDICATE_CONFIG.equals(entry.getKey())
+ || TransformationStage.NEGATE_CONFIG.equals(entry.getKey())) {
+ log.warn("Transformer config {} is masked by implicit config of that name",
+ entry.getKey());
+ return false;
+ } else {
+ return true;
+ }
+ });
}
@Override
@@ -464,87 +405,6 @@ public class ConnectorConfig extends AbstractConfig {
return newDef;
}
- private static <T> ConverterDefaults converterDefaults(
- Plugins plugins,
- String connectorConverterConfig,
- String workerConverterConfig,
- String workerConverterVersionConfig,
- Map<String, String> connectorProps,
- WorkerConfig workerConfig,
- Class<T> converterType
- ) {
- /*
- if a converter is specified in the connector config it overrides the worker config for the corresponding converter
- otherwise the worker config is used, hence if the converter is not provided in the connector config, the default
- is the one provided in the worker config
-
- for converters which version is used depends on a several factors with multi-versioning support
- A. If the converter class is provided as part of the connector properties
- 1. if the version is not provided,
- - if the converter is packaged with the connector then, the packaged version is used
- - if the converter is not packaged with the connector, the latest version is used
- 2. if the version is provided, the provided version is used
- B. If the converter class is not provided as part of the connector properties, but provided as part of the worker properties
- 1. if the version is not provided, the latest version is used
- 2. if the version is provided, the provided version is used
- C. If the converter class is not provided as part of the connector properties and not provided as part of the worker properties,
- the converter to use is unknown hence no default version can be determined (null)
-
- Note: Connect when using service loading has an issue outlined in KAFKA-18119. The issue means that the above
- logic does not hold currently for clusters using service loading when converters are defined in the connector.
- However, the logic to determine the default should ideally follow the one outlined above, and the code here
- should still show the correct default version regardless of the bug.
- */
- final String connectorConverter = connectorProps.get(connectorConverterConfig);
- // since header converter defines a default in the worker config we need to handle it separately
- final String workerConverter = workerConverterConfig.equals(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG) ?
- workerConfig.getClass(workerConverterConfig).getName() : workerConfig.originalsStrings().get(workerConverterConfig);
- final String connectorClass = connectorProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
- final String connectorVersion = connectorProps.get(ConnectorConfig.CONNECTOR_VERSION);
- String type = null;
- if (connectorClass == null || (connectorConverter == null && workerConverter == null)) {
- return new ConverterDefaults(null, null);
- }
- // update the default of connector converter based on if the worker converter is provided
- type = workerConverter;
-
- String version = null;
- if (connectorConverter != null) {
- version = fetchPluginVersion(plugins, connectorConverter, connectorVersion, connectorConverter);
- } else {
- version = workerConfig.originalsStrings().get(workerConverterVersionConfig);
- if (version == null) {
- version = plugins.latestVersion(workerConverter);
- }
- }
- return new ConverterDefaults(type, version);
- }
-
- private static void updateKeyDefault(ConfigDef configDef, String versionConfigKey, String versionDefault) {
- ConfigDef.ConfigKey key = configDef.configKeys().get(versionConfigKey);
- if (key == null) {
- return;
- }
- configDef.configKeys().put(versionConfigKey, new ConfigDef.ConfigKey(
- versionConfigKey, key.type, versionDefault, key.validator, key.importance, key.documentation, key.group, key.orderInGroup, key.width, key.displayName, key.dependents, key.recommender, false
- ));
- }
-
- @SuppressWarnings("unchecked")
- private static <T> String fetchPluginVersion(Plugins plugins, String connectorClass, String connectorVersion, String pluginName) {
- if (pluginName == null) {
- return null;
- }
- try {
- VersionRange range = PluginUtils.connectorVersionRequirement(connectorVersion);
- return plugins.pluginVersion(pluginName, plugins.pluginLoader(connectorClass, range));
- } catch (InvalidVersionSpecificationException | VersionedPluginLoadingException e) {
- // these errors should be captured in other places, so we can ignore them here
- log.warn("Failed to determine default plugin version for {}", connectorClass, e);
- }
- return null;
- }
-
/**
* An abstraction over "enrichable plugins" ({@link Transformation}s and {@link Predicate}s) used for computing the
* contribution to a Connectors ConfigDef.
@@ -595,14 +455,14 @@ public class ConnectorConfig extends AbstractConfig {
final String typeConfig = prefix + "type";
final ConfigDef.Validator typeValidator = ConfigDef.LambdaValidator.with(
- (String name, Object value) -> {
- validateProps(prefix);
- // The value will be null if the class couldn't be found; no point in performing follow-up validation
- if (value != null) {
- getConfigDefFromConfigProvidingClass(typeConfig, (Class<?>) value);
- }
- },
- () -> "valid configs for " + alias + " " + aliasKind.toLowerCase(Locale.ENGLISH));
+ (String name, Object value) -> {
+ validateProps(prefix);
+ // The value will be null if the class couldn't be found; no point in performing follow-up validation
+ if (value != null) {
+ getConfigDefFromConfigProvidingClass(typeConfig, (Class<?>) value);
+ }
+ },
+ () -> "valid configs for " + alias + " " + aliasKind.toLowerCase(Locale.ENGLISH));
newDef.define(typeConfig, Type.CLASS, ConfigDef.NO_DEFAULT_VALUE, typeValidator, Importance.HIGH,
"Class for the '" + alias + "' " + aliasKind.toLowerCase(Locale.ENGLISH) + ".", group, orderInGroup++, Width.LONG,
baseClass.getSimpleName() + " type for " + alias,
@@ -615,8 +475,7 @@ public class ConnectorConfig extends AbstractConfig {
}
/** Subclasses can add extra validation of the {@link #props}. */
- protected void validateProps(String prefix) {
- }
+ protected void validateProps(String prefix) { }
/**
* Populates the ConfigDef according to the configs returned from {@code configs()} method of class
@@ -627,6 +486,7 @@ public class ConnectorConfig extends AbstractConfig {
try {
configDefsForClass(typeConfig)
.forEach(entry -> configDef.define(entry.getValue()));
+
} catch (ConfigException e) {
if (requireFullConfig) {
throw e;
@@ -673,10 +533,10 @@ public class ConnectorConfig extends AbstractConfig {
ConfigDef configDef = config(pluginInstance);
if (null == configDef) {
throw new ConnectException(
- String.format(
- "%s.config() must return a ConfigDef that is not null.",
- cls.getName()
- )
+ String.format(
+ "%s.config() must return a ConfigDef that is not null.",
+ cls.getName()
+ )
);
}
return configDef;
@@ -716,34 +576,4 @@ public class ConnectorConfig extends AbstractConfig {
}
}
- private static class ConverterDefaults {
- private final String type;
- private final String version;
-
- public ConverterDefaults(String type, String version) {
- this.type = type;
- this.version = version;
- }
-
- public String type() {
- return type;
- }
-
- public String version() {
- return version;
- }
- }
-
- public static class PluginVersionValidator implements ConfigDef.Validator {
-
- @Override
- public void ensureValid(String name, Object value) {
-
- try {
- PluginUtils.connectorVersionRequirement((String) value);
- } catch (InvalidVersionSpecificationException e) {
- throw new ConfigException(name, value, e.getMessage());
- }
- }
- }
}
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 859e3f2728e..52be401bbfa 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
@@ -32,8 +32,6 @@ import org.apache.kafka.connect.storage.StatusBackingStore;
import org.apache.kafka.connect.util.Callback;
import org.apache.kafka.connect.util.ConnectorTaskId;
-import org.apache.maven.artifact.versioning.VersionRange;
-
import java.util.Collection;
import java.util.List;
import java.util.Map;
@@ -324,8 +322,6 @@ public interface Herder {
*/
List<ConfigKeyInfo> connectorPluginConfig(String pluginName);
- List<ConfigKeyInfo> connectorPluginConfig(String pluginName, VersionRange version);
-
/**
* Get the current offsets for a connector.
* @param connName the name of the connector whose offsets are to be retrieved
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SinkConnectorConfig.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SinkConnectorConfig.java
index 4584255e231..2ab7dfa0897 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SinkConnectorConfig.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SinkConnectorConfig.java
@@ -73,29 +73,19 @@ public class SinkConnectorConfig extends ConnectorConfig {
"keys, all error context header keys will start with <code>__connect.errors.</code>";
private static final String DLQ_CONTEXT_HEADERS_ENABLE_DISPLAY = "Enable Error Context Headers";
- private static ConfigDef configDef(ConfigDef baseConfigs) {
- return baseConfigs
- .define(TOPICS_CONFIG, ConfigDef.Type.LIST, TOPICS_DEFAULT, ConfigDef.Importance.HIGH, TOPICS_DOC, COMMON_GROUP, 4, ConfigDef.Width.LONG, TOPICS_DISPLAY)
- .define(TOPICS_REGEX_CONFIG, ConfigDef.Type.STRING, TOPICS_REGEX_DEFAULT, new RegexValidator(), ConfigDef.Importance.HIGH, TOPICS_REGEX_DOC, COMMON_GROUP, 4, ConfigDef.Width.LONG, TOPICS_REGEX_DISPLAY)
- .define(DLQ_TOPIC_NAME_CONFIG, ConfigDef.Type.STRING, DLQ_TOPIC_DEFAULT, Importance.MEDIUM, DLQ_TOPIC_NAME_DOC, ERROR_GROUP, 6, ConfigDef.Width.MEDIUM, DLQ_TOPIC_DISPLAY)
- .define(DLQ_TOPIC_REPLICATION_FACTOR_CONFIG, ConfigDef.Type.SHORT, DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DEFAULT, Importance.MEDIUM, DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DOC, ERROR_GROUP, 7, ConfigDef.Width.MEDIUM, DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DISPLAY)
- .define(DLQ_CONTEXT_HEADERS_ENABLE_CONFIG, ConfigDef.Type.BOOLEAN, DLQ_CONTEXT_HEADERS_ENABLE_DEFAULT, Importance.MEDIUM, DLQ_CONTEXT_HEADERS_ENABLE_DOC, ERROR_GROUP, 8, ConfigDef.Width.MEDIUM, DLQ_CONTEXT_HEADERS_ENABLE_DISPLAY);
- }
+ static final ConfigDef CONFIG = ConnectorConfig.configDef()
+ .define(TOPICS_CONFIG, ConfigDef.Type.LIST, TOPICS_DEFAULT, ConfigDef.Importance.HIGH, TOPICS_DOC, COMMON_GROUP, 4, ConfigDef.Width.LONG, TOPICS_DISPLAY)
+ .define(TOPICS_REGEX_CONFIG, ConfigDef.Type.STRING, TOPICS_REGEX_DEFAULT, new RegexValidator(), ConfigDef.Importance.HIGH, TOPICS_REGEX_DOC, COMMON_GROUP, 4, ConfigDef.Width.LONG, TOPICS_REGEX_DISPLAY)
+ .define(DLQ_TOPIC_NAME_CONFIG, ConfigDef.Type.STRING, DLQ_TOPIC_DEFAULT, Importance.MEDIUM, DLQ_TOPIC_NAME_DOC, ERROR_GROUP, 6, ConfigDef.Width.MEDIUM, DLQ_TOPIC_DISPLAY)
+ .define(DLQ_TOPIC_REPLICATION_FACTOR_CONFIG, ConfigDef.Type.SHORT, DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DEFAULT, Importance.MEDIUM, DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DOC, ERROR_GROUP, 7, ConfigDef.Width.MEDIUM, DLQ_TOPIC_REPLICATION_FACTOR_CONFIG_DISPLAY)
+ .define(DLQ_CONTEXT_HEADERS_ENABLE_CONFIG, ConfigDef.Type.BOOLEAN, DLQ_CONTEXT_HEADERS_ENABLE_DEFAULT, Importance.MEDIUM, DLQ_CONTEXT_HEADERS_ENABLE_DOC, ERROR_GROUP, 8, ConfigDef.Width.MEDIUM, DLQ_CONTEXT_HEADERS_ENABLE_DISPLAY);
public static ConfigDef configDef() {
- return configDef(ConnectorConfig.configDef());
- }
-
- public static ConfigDef enrichedConfigDef(Plugins plugins, Map<String, String> connProps, WorkerConfig workerConfig) {
- return configDef(ConnectorConfig.enrichedConfigDef(plugins, connProps, workerConfig));
- }
-
- public static ConfigDef enrichedConfigDef(Plugins plugins, String connectorClass) {
- return configDef(ConnectorConfig.enrichedConfigDef(plugins, connectorClass));
+ return CONFIG;
}
public SinkConnectorConfig(Plugins plugins, Map<String, String> props) {
- super(plugins, configDef(), props);
+ super(plugins, CONFIG, props);
}
/**
@@ -216,6 +206,6 @@ public class SinkConnectorConfig extends ConnectorConfig {
}
public static void main(String[] args) {
- System.out.println(configDef().toHtml(4, config -> "sinkconnectorconfigs_" + config));
+ System.out.println(CONFIG.toHtml(4, config -> "sinkconnectorconfigs_" + config));
}
}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceConnectorConfig.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceConnectorConfig.java
index effa3135376..bc797563b10 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceConnectorConfig.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceConnectorConfig.java
@@ -125,10 +125,10 @@ public final class SourceConnectorConfig extends ConnectorConfig {
private final EnrichedSourceConnectorConfig enrichedSourceConfig;
private final String offsetsTopic;
- private static ConfigDef configDef(ConfigDef baseConfigDef) {
+ public static ConfigDef configDef() {
ConfigDef.Validator atLeastZero = ConfigDef.Range.atLeast(0);
int orderInGroup = 0;
- return new ConfigDef(baseConfigDef)
+ return new ConfigDef(ConnectorConfig.configDef())
.define(
TOPIC_CREATION_GROUPS_CONFIG,
ConfigDef.Type.LIST,
@@ -203,18 +203,6 @@ public final class SourceConnectorConfig extends ConnectorConfig {
OFFSETS_TOPIC_DISPLAY);
}
- public static ConfigDef configDef() {
- return configDef(ConnectorConfig.configDef());
- }
-
- public static ConfigDef enrichedConfigDef(Plugins plugins, Map<String, String> connProps, WorkerConfig workerConfig) {
- return configDef(ConnectorConfig.enrichedConfigDef(plugins, connProps, workerConfig));
- }
-
- public static ConfigDef enrichedConfigDef(Plugins plugins, String connectorClass) {
- return configDef(ConnectorConfig.enrichedConfigDef(plugins, connectorClass));
- }
-
public static ConfigDef embedDefaultGroup(ConfigDef baseConfigDef) {
String defaultGroup = "default";
ConfigDef newDefaultDef = new ConfigDef(baseConfigDef);
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
index 1f97a907b64..591e9816a7a 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
@@ -278,10 +278,6 @@ public final class Worker {
ThreadUtils.shutdownExecutorServiceQuietly(executor, EXECUTOR_SHUTDOWN_TERMINATION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
}
- public WorkerConfig config() {
- return config;
- }
-
/**
* Start a connector managed by this worker.
*
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java
index a68cdb4ea03..ca188ffd97a 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java
@@ -70,8 +70,6 @@ public class WorkerConfig extends AbstractConfig {
public static final String CLIENT_DNS_LOOKUP_CONFIG = CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG;
public static final String CLIENT_DNS_LOOKUP_DOC = CommonClientConfigs.CLIENT_DNS_LOOKUP_DOC;
- public static final String PLUGIN_VERSION_SUFFIX = "plugin.version";
-
public static final String KEY_CONVERTER_CLASS_CONFIG = "key.converter";
public static final String KEY_CONVERTER_CLASS_DOC =
"Converter class used to convert between Kafka Connect format and the serialized form that is written to Kafka." +
@@ -79,10 +77,6 @@ public class WorkerConfig extends AbstractConfig {
" independent of connectors it allows any connector to work with any serialization format." +
" Examples of common formats include JSON and Avro.";
- public static final String KEY_CONVERTER_VERSION = "key.converter." + PLUGIN_VERSION_SUFFIX;
- public static final String KEY_CONVERTER_VERSION_DEFAULT = null;
- public static final String KEY_CONVERTER_VERSION_DOC = "Version of the key converter.";
-
public static final String VALUE_CONVERTER_CLASS_CONFIG = "value.converter";
public static final String VALUE_CONVERTER_CLASS_DOC =
"Converter class used to convert between Kafka Connect format and the serialized form that is written to Kafka." +
@@ -90,10 +84,6 @@ public class WorkerConfig extends AbstractConfig {
" independent of connectors it allows any connector to work with any serialization format." +
" Examples of common formats include JSON and Avro.";
- public static final String VALUE_CONVERTER_VERSION = "value.converter." + PLUGIN_VERSION_SUFFIX;
- public static final String VALUE_CONVERTER_VERSION_DEFAULT = null;
- public static final String VALUE_CONVERTER_VERSION_DOC = "Version of the value converter.";
-
public static final String HEADER_CONVERTER_CLASS_CONFIG = "header.converter";
public static final String HEADER_CONVERTER_CLASS_DOC =
"HeaderConverter class used to convert between Kafka Connect format and the serialized form that is written to Kafka." +
@@ -103,10 +93,6 @@ public class WorkerConfig extends AbstractConfig {
" header values to strings and deserialize them by inferring the schemas.";
public static final String HEADER_CONVERTER_CLASS_DEFAULT = SimpleHeaderConverter.class.getName();
- public static final String HEADER_CONVERTER_VERSION = "header.converter." + PLUGIN_VERSION_SUFFIX;
- public static final String HEADER_CONVERTER_VERSION_DEFAULT = null;
- public static final String HEADER_CONVERTER_VERSION_DOC = "Version of the header converter.";
-
public static final String TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_CONFIG
= "task.shutdown.graceful.timeout.ms";
private static final String TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DOC =
@@ -214,12 +200,8 @@ public class WorkerConfig extends AbstractConfig {
CLIENT_DNS_LOOKUP_DOC)
.define(KEY_CONVERTER_CLASS_CONFIG, Type.CLASS,
Importance.HIGH, KEY_CONVERTER_CLASS_DOC)
- .define(KEY_CONVERTER_VERSION, Type.STRING,
- KEY_CONVERTER_VERSION_DEFAULT, Importance.LOW, KEY_CONVERTER_VERSION_DOC)
.define(VALUE_CONVERTER_CLASS_CONFIG, Type.CLASS,
Importance.HIGH, VALUE_CONVERTER_CLASS_DOC)
- .define(VALUE_CONVERTER_VERSION, Type.STRING,
- VALUE_CONVERTER_VERSION_DEFAULT, Importance.LOW, VALUE_CONVERTER_VERSION_DOC)
.define(TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_CONFIG, Type.LONG,
TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DEFAULT, Importance.LOW,
TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DOC)
@@ -255,8 +237,6 @@ public class WorkerConfig extends AbstractConfig {
.define(HEADER_CONVERTER_CLASS_CONFIG, Type.CLASS,
HEADER_CONVERTER_CLASS_DEFAULT,
Importance.LOW, HEADER_CONVERTER_CLASS_DOC)
- .define(HEADER_CONVERTER_VERSION, Type.STRING,
- HEADER_CONVERTER_VERSION_DEFAULT, Importance.LOW, HEADER_CONVERTER_VERSION_DOC)
.define(CONFIG_PROVIDERS_CONFIG, Type.LIST,
Collections.emptyList(),
Importance.LOW, CONFIG_PROVIDERS_DOC)
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 6a850439287..fdbadef7b69 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
@@ -16,23 +16,17 @@
*/
package org.apache.kafka.connect.runtime.isolation;
-import org.apache.maven.artifact.versioning.ArtifactVersion;
-import org.apache.maven.artifact.versioning.DefaultArtifactVersion;
-import org.apache.maven.artifact.versioning.VersionRange;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.URL;
import java.net.URLClassLoader;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import java.util.stream.Collectors;
/**
* A custom classloader dedicated to loading Connect plugin classes in classloading isolation.
@@ -75,125 +69,36 @@ public class DelegatingClassLoader extends URLClassLoader {
/**
* Retrieve the PluginClassLoader associated with a plugin class
- *
* @param name The fully qualified class name of the plugin
* @return the PluginClassLoader that should be used to load this, or null if the plugin is not isolated.
*/
// VisibleForTesting
- PluginClassLoader pluginClassLoader(String name, VersionRange range) {
+ PluginClassLoader pluginClassLoader(String name) {
if (!PluginUtils.shouldLoadInIsolation(name)) {
return null;
}
-
SortedMap<PluginDesc<?>, ClassLoader> inner = pluginLoaders.get(name);
if (inner == null) {
return null;
}
-
-
- ClassLoader pluginLoader = findPluginLoader(inner, name, range);
+ ClassLoader pluginLoader = inner.get(inner.lastKey());
return pluginLoader instanceof PluginClassLoader
- ? (PluginClassLoader) pluginLoader
- : null;
- }
-
- PluginClassLoader pluginClassLoader(String name) {
- return pluginClassLoader(name, null);
+ ? (PluginClassLoader) pluginLoader
+ : null;
}
- ClassLoader loader(String classOrAlias, VersionRange range) {
- String fullName = aliases.getOrDefault(classOrAlias, classOrAlias);
- ClassLoader classLoader = pluginClassLoader(fullName, range);
- if (classLoader == null) {
- classLoader = this;
- }
+ ClassLoader connectorLoader(String connectorClassOrAlias) {
+ String fullName = aliases.getOrDefault(connectorClassOrAlias, connectorClassOrAlias);
+ ClassLoader classLoader = pluginClassLoader(fullName);
+ if (classLoader == null) classLoader = this;
log.debug(
- "Got plugin class loader: '{}' for connector: {}",
- classLoader,
- classOrAlias
+ "Getting plugin class loader: '{}' for connector: {}",
+ classLoader,
+ connectorClassOrAlias
);
return classLoader;
}
- ClassLoader loader(String classOrAlias) {
- return loader(classOrAlias, null);
- }
-
- ClassLoader connectorLoader(String connectorClassOrAlias) {
- return loader(connectorClassOrAlias);
- }
-
- String resolveFullClassName(String classOrAlias) {
- return aliases.getOrDefault(classOrAlias, classOrAlias);
- }
-
- String latestVersion(String classOrAlias) {
- if (classOrAlias == null) {
- return null;
- }
- String fullName = aliases.getOrDefault(classOrAlias, classOrAlias);
- SortedMap<PluginDesc<?>, ClassLoader> inner = pluginLoaders.get(fullName);
- if (inner == null) {
- return null;
- }
- return inner.lastKey().version();
- }
-
- String versionInLocation(String classOrAlias, String location) {
- if (classOrAlias == null) {
- return null;
- }
- String fullName = aliases.getOrDefault(classOrAlias, classOrAlias);
- SortedMap<PluginDesc<?>, ClassLoader> inner = pluginLoaders.get(fullName);
- if (inner == null) {
- return null;
- }
- for (Map.Entry<PluginDesc<?>, ClassLoader> entry : inner.entrySet()) {
- if (entry.getKey().location().equals(location)) {
- return entry.getKey().version();
- }
- }
- return null;
- }
-
- private ClassLoader findPluginLoader(
- SortedMap<PluginDesc<?>, ClassLoader> loaders,
- String pluginName,
- VersionRange range
- ) {
-
- if (range != null) {
-
- if (null != range.getRecommendedVersion()) {
- throw new VersionedPluginLoadingException(String.format("A soft version range is not supported for plugin loading, "
- + "this is an internal error as connect should automatically convert soft ranges to hard ranges. "
- + "Provided soft version: %s ", range));
- }
-
- ArtifactVersion version = null;
- ClassLoader loader = null;
- for (Map.Entry<PluginDesc<?>, ClassLoader> entry : loaders.entrySet()) {
- // the entries should be in sorted order of versions so this should end up picking the latest version which matches the range
- if (range.containsVersion(entry.getKey().encodedVersion())) {
- loader = entry.getValue();
- }
- }
-
- if (loader == null) {
- List<String> availableVersions = loaders.keySet().stream().map(PluginDesc::version).collect(Collectors.toList());
- throw new VersionedPluginLoadingException(String.format(
- "Plugin %s not found that matches the version range %s, available versions: %s",
- pluginName,
- range,
- availableVersions
- ), availableVersions);
- }
- return loader;
- }
-
- return loaders.get(loaders.lastKey());
- }
-
public void installDiscoveredPlugins(PluginScanResult scanResult) {
pluginLoaders.putAll(computePluginLoaders(scanResult));
for (String pluginClassName : pluginLoaders.keySet()) {
@@ -207,76 +112,21 @@ public class DelegatingClassLoader extends URLClassLoader {
@Override
protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException {
- return loadVersionedPluginClass(name, null, resolve);
- }
-
- protected Class<?> loadVersionedPluginClass(
- String name,
- VersionRange range,
- boolean resolve
- ) throws VersionedPluginLoadingException, ClassNotFoundException {
-
String fullName = aliases.getOrDefault(name, name);
- PluginClassLoader pluginLoader = pluginClassLoader(fullName, range);
- Class<?> plugin;
+ PluginClassLoader pluginLoader = pluginClassLoader(fullName);
if (pluginLoader != null) {
- log.trace("Retrieving loaded class '{}' from '{}'", name, pluginLoader);
- plugin = pluginLoader.loadClass(fullName, resolve);
- } else {
- plugin = super.loadClass(fullName, resolve);
- if (range == null) {
- return plugin;
- }
- verifyClasspathVersionedPlugin(name, plugin, range);
+ log.trace("Retrieving loaded class '{}' from '{}'", fullName, pluginLoader);
+ return pluginLoader.loadClass(fullName, resolve);
}
- return plugin;
- }
-
- private void verifyClasspathVersionedPlugin(String name, Class<?> plugin, VersionRange range) throws VersionedPluginLoadingException {
- String pluginVersion;
- SortedMap<PluginDesc<?>, ClassLoader> scannedPlugin = pluginLoaders.get(name);
-
- if (scannedPlugin == null) {
- throw new VersionedPluginLoadingException(String.format(
- "Plugin %s is not part of Connect's plugin loading mechanism (ClassPath or Plugin Path)",
- name
- ));
- }
-
- // if a plugin implements two interfaces (like JsonConverter implements both converter and header converter)
- // it will have two entries under classpath, one for each scan. Hence, we count distinct by version.
- List<String> classpathPlugins = scannedPlugin.keySet().stream()
- .filter(pluginDesc -> pluginDesc.location().equals("classpath"))
- .map(PluginDesc::version)
- .distinct()
- .collect(Collectors.toList());
- if (classpathPlugins.size() > 1) {
- throw new VersionedPluginLoadingException(String.format(
- "Plugin %s has multiple versions specified in class path, "
- + "only one version is allowed in class path for loading a plugin with version range",
- name
- ));
- } else if (classpathPlugins.isEmpty()) {
- throw new VersionedPluginLoadingException("Invalid plugin found in classpath");
- } else {
- pluginVersion = classpathPlugins.get(0);
- if (!range.containsVersion(new DefaultArtifactVersion(pluginVersion))) {
- throw new VersionedPluginLoadingException(String.format(
- "Plugin %s has version %s which does not match the required version range %s",
- name,
- pluginVersion,
- range
- ), Collections.singletonList(pluginVersion));
- }
- }
+ return super.loadClass(fullName, resolve);
}
private static Map<String, SortedMap<PluginDesc<?>, ClassLoader>> computePluginLoaders(PluginScanResult plugins) {
Map<String, SortedMap<PluginDesc<?>, ClassLoader>> pluginLoaders = new HashMap<>();
plugins.forEach(pluginDesc ->
- pluginLoaders.computeIfAbsent(pluginDesc.className(), k -> new TreeMap<>())
- .put(pluginDesc, pluginDesc.loader()));
+ pluginLoaders.computeIfAbsent(pluginDesc.className(), k -> new TreeMap<>())
+ .put(pluginDesc, pluginDesc.loader()));
return pluginLoaders;
}
}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginDesc.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginDesc.java
index 1fecb59ddb5..a58aef7ceca 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginDesc.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginDesc.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.connect.runtime.isolation;
-import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.maven.artifact.versioning.DefaultArtifactVersion;
@@ -61,11 +60,6 @@ public class PluginDesc<T> implements Comparable<PluginDesc<?>> {
'}';
}
- @JsonIgnore
- public DefaultArtifactVersion encodedVersion() {
- return encodedVersion;
- }
-
public Class<? extends T> pluginClass() {
return klass;
}
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 ae6d3ba3a1c..932e87395f7 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
@@ -16,8 +16,6 @@
*/
package org.apache.kafka.connect.runtime.isolation;
-import org.apache.maven.artifact.versioning.InvalidVersionSpecificationException;
-import org.apache.maven.artifact.versioning.VersionRange;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -471,7 +469,7 @@ public class PluginUtils {
}
return distinctUrls(urls);
}
-
+
private static Collection<URL> forClassLoader(ClassLoader classLoader) {
final Collection<URL> result = new ArrayList<>();
while (classLoader != null) {
@@ -485,7 +483,7 @@ public class PluginUtils {
}
return distinctUrls(result);
}
-
+
private static Collection<URL> distinctUrls(Collection<URL> urls) {
Map<String, URL> distinct = new HashMap<>(urls.size());
for (URL url : urls) {
@@ -493,21 +491,4 @@ public class PluginUtils {
}
return distinct.values();
}
-
- public static VersionRange connectorVersionRequirement(String version) throws InvalidVersionSpecificationException {
- if (version == null || version.equals("latest")) {
- return null;
- }
- version = version.trim();
-
- // check first if the given version is valid
- VersionRange range = VersionRange.createFromVersionSpec(version);
-
- if (range.hasRestrictions()) {
- return range;
- }
- // now if the version is not enclosed we consider it as a hard requirement and enclose it in []
- version = "[" + version + "]";
- return VersionRange.createFromVersionSpec(version);
- }
}
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 b8027d71d56..816f870157e 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
@@ -35,8 +35,6 @@ import org.apache.kafka.connect.storage.HeaderConverter;
import org.apache.kafka.connect.transforms.Transformation;
import org.apache.kafka.connect.transforms.predicates.Predicate;
-import org.apache.maven.artifact.versioning.InvalidVersionSpecificationException;
-import org.apache.maven.artifact.versioning.VersionRange;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,7 +48,6 @@ import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
-import java.util.function.Function;
import java.util.stream.Collectors;
public class Plugins {
@@ -171,20 +168,11 @@ public class Plugins {
String classOrAlias,
Class<U> pluginClass
) throws ClassNotFoundException {
- return pluginClass(loader, classOrAlias, pluginClass, null);
- }
-
- @SuppressWarnings("unchecked")
- protected static <U> Class<? extends U> pluginClass(
- DelegatingClassLoader loader,
- String classOrAlias,
- Class<U> pluginClass,
- VersionRange range
- ) throws VersionedPluginLoadingException, ClassNotFoundException {
- Class<?> klass = loader.loadVersionedPluginClass(classOrAlias, range, false);
+ Class<?> klass = loader.loadClass(classOrAlias, false);
if (pluginClass.isAssignableFrom(klass)) {
return (Class<? extends U>) klass;
}
+
throw new ClassNotFoundException(
"Requested class: "
+ classOrAlias
@@ -196,10 +184,6 @@ public class Plugins {
return pluginClass(delegatingLoader, classOrAlias, Object.class);
}
- public Class<?> pluginClass(String classOrAlias, VersionRange range) throws VersionedPluginLoadingException, ClassNotFoundException {
- return pluginClass(delegatingLoader, classOrAlias, Object.class, range);
- }
-
public static ClassLoader compareAndSwapLoaders(ClassLoader loader) {
ClassLoader current = Thread.currentThread().getContextClassLoader();
if (!current.equals(loader)) {
@@ -256,44 +240,14 @@ public class Plugins {
};
}
- public Function<ClassLoader, LoaderSwap> safeLoaderSwapper() {
- return loader -> {
- if (!(loader instanceof PluginClassLoader)) {
- loader = delegatingLoader;
- }
- return withClassLoader(loader);
- };
- }
-
- public String latestVersion(String classOrAlias) {
- return delegatingLoader.latestVersion(classOrAlias);
- }
-
- public String pluginVersion(String classOrAlias, ClassLoader sourceLoader) {
- if (!(sourceLoader instanceof PluginClassLoader)) {
- return latestVersion(classOrAlias);
- }
- return delegatingLoader.versionInLocation(classOrAlias, ((PluginClassLoader) sourceLoader).location());
- }
-
public DelegatingClassLoader delegatingLoader() {
return delegatingLoader;
}
- // kept for compatibility
public ClassLoader connectorLoader(String connectorClassOrAlias) {
- return delegatingLoader.loader(connectorClassOrAlias);
+ return delegatingLoader.connectorLoader(connectorClassOrAlias);
}
- public ClassLoader pluginLoader(String classOrAlias, VersionRange range) {
- return delegatingLoader.loader(classOrAlias, range);
- }
-
- public ClassLoader pluginLoader(String classOrAlias) {
- return delegatingLoader.loader(classOrAlias);
- }
-
-
@SuppressWarnings({"unchecked", "rawtypes"})
public Set<PluginDesc<Connector>> connectors() {
Set<PluginDesc<Connector>> connectors = new TreeSet<>((Set) sinkConnectors());
@@ -305,96 +259,48 @@ public class Plugins {
return scanResult.sinkConnectors();
}
- Set<PluginDesc<SinkConnector>> sinkConnectors(String connectorClassOrAlias) {
- return pluginsOfClass(connectorClassOrAlias, scanResult.sinkConnectors());
- }
-
public Set<PluginDesc<SourceConnector>> sourceConnectors() {
return scanResult.sourceConnectors();
}
- Set<PluginDesc<SourceConnector>> sourceConnectors(String connectorClassOrAlias) {
- return pluginsOfClass(connectorClassOrAlias, scanResult.sourceConnectors());
- }
-
public Set<PluginDesc<Converter>> converters() {
return scanResult.converters();
}
- Set<PluginDesc<Converter>> converters(String converterClassOrAlias) {
- return pluginsOfClass(converterClassOrAlias, scanResult.converters());
- }
-
public Set<PluginDesc<HeaderConverter>> headerConverters() {
return scanResult.headerConverters();
}
- Set<PluginDesc<HeaderConverter>> headerConverters(String headerConverterClassOrAlias) {
- return pluginsOfClass(headerConverterClassOrAlias, scanResult.headerConverters());
- }
-
public Set<PluginDesc<Transformation<?>>> transformations() {
return scanResult.transformations();
}
- Set<PluginDesc<Transformation<?>>> transformations(String transformationClassOrAlias) {
- return pluginsOfClass(transformationClassOrAlias, scanResult.transformations());
- }
-
public Set<PluginDesc<Predicate<?>>> predicates() {
return scanResult.predicates();
}
- Set<PluginDesc<Predicate<?>>> predicates(String predicateClassOrAlias) {
- return pluginsOfClass(predicateClassOrAlias, scanResult.predicates());
- }
-
public Set<PluginDesc<ConnectorClientConfigOverridePolicy>> connectorClientConfigPolicies() {
return scanResult.connectorClientConfigPolicies();
}
- private <T> Set<PluginDesc<T>> pluginsOfClass(String classNameOrAlias, Set<PluginDesc<T>> allPluginsOfType) {
- String className = delegatingLoader.resolveFullClassName(classNameOrAlias);
- Set<PluginDesc<T>> plugins = new TreeSet<>();
- for (PluginDesc<T> desc : allPluginsOfType) {
- if (desc.className().equals(className)) {
- plugins.add(desc);
- }
- }
- return plugins;
- }
-
public Object newPlugin(String classOrAlias) throws ClassNotFoundException {
Class<?> klass = pluginClass(delegatingLoader, classOrAlias, Object.class);
return newPlugin(klass);
}
- public Object newPlugin(String classOrAlias, VersionRange range) throws VersionedPluginLoadingException, ClassNotFoundException {
- Class<?> klass = pluginClass(delegatingLoader, classOrAlias, Object.class, range);
- return newPlugin(klass);
- }
-
- public Object newPlugin(String classOrAlias, VersionRange range, ClassLoader sourceLoader) throws ClassNotFoundException {
- if (range == null && sourceLoader instanceof PluginClassLoader) {
- sourceLoader.loadClass(classOrAlias);
- }
- return newPlugin(classOrAlias, range);
- }
-
public Connector newConnector(String connectorClassOrAlias) {
Class<? extends Connector> klass = connectorClass(connectorClassOrAlias);
return newPlugin(klass);
}
- public Connector newConnector(String connectorClassOrAlias, VersionRange range) throws VersionedPluginLoadingException {
- Class<? extends Connector> klass = connectorClass(connectorClassOrAlias, range);
- return newPlugin(klass);
- }
-
- public Class<? extends Connector> connectorClass(String connectorClassOrAlias, VersionRange range) throws VersionedPluginLoadingException {
+ public Class<? extends Connector> connectorClass(String connectorClassOrAlias) {
Class<? extends Connector> klass;
try {
- klass = pluginClass(delegatingLoader, connectorClassOrAlias, Connector.class, range);
+ klass = pluginClass(
+ delegatingLoader,
+ connectorClassOrAlias,
+ Connector.class
+ );
} catch (ClassNotFoundException e) {
List<PluginDesc<? extends Connector>> matches = new ArrayList<>();
Set<PluginDesc<Connector>> connectors = connectors();
@@ -430,10 +336,6 @@ public class Plugins {
return klass;
}
- public Class<? extends Connector> connectorClass(String connectorClassOrAlias) {
- return connectorClass(connectorClassOrAlias, null);
- }
-
public Task newTask(Class<? extends Task> taskClass) {
return newPlugin(taskClass);
}
@@ -448,49 +350,54 @@ public class Plugins {
* @throws ConnectException if the {@link Converter} implementation class could not be found
*/
public Converter newConverter(AbstractConfig config, String classPropertyName, ClassLoaderUsage classLoaderUsage) {
- return newConverter(config, classPropertyName, null, classLoaderUsage);
- }
-
- /**
- * Used to get a versioned converter. If the version is specified, it will always use the plugins classloader.
- *
- * @param config the configuration containing the {@link Converter}'s configuration; may not be null
- * @param classPropertyName the name of the property that contains the name of the {@link Converter} class; may not be null
- * @param versionPropertyName the name of the property that contains the version of the {@link Converter} class; may not be null
- * @return the instantiated and configured {@link Converter}; null if the configuration did not define the specified property
- * @throws ConnectException if the {@link Converter} implementation class could not be found,
- * @throws VersionedPluginLoadingException if the version requested is not found
- */
- public Converter newConverter(AbstractConfig config, String classPropertyName, String versionPropertyName) {
- ClassLoaderUsage classLoader = config.getString(versionPropertyName) == null ? ClassLoaderUsage.CURRENT_CLASSLOADER : ClassLoaderUsage.PLUGINS;
- return newConverter(config, classPropertyName, versionPropertyName, classLoader);
- }
-
- private Converter newConverter(AbstractConfig config, String classPropertyName, String versionPropertyName, ClassLoaderUsage classLoaderUsage) {
if (!config.originals().containsKey(classPropertyName)) {
// This configuration does not define the converter via the specified property name
return null;
}
+
+ Class<? extends Converter> klass = null;
+ switch (classLoaderUsage) {
+ case CURRENT_CLASSLOADER:
+ // Attempt to load first with the current classloader, and plugins as a fallback.
+ // Note: we can't use config.getConfiguredInstance because Converter doesn't implement Configurable, and even if it did
+ // we have to remove the property prefixes before calling config(...) and we still always want to call Converter.config.
+ klass = pluginClassFromConfig(config, classPropertyName, Converter.class, scanResult.converters());
+ break;
+ case PLUGINS:
+ // Attempt to load with the plugin class loader, which uses the current classloader as a fallback
+ String converterClassOrAlias = config.getClass(classPropertyName).getName();
+ try {
+ klass = pluginClass(delegatingLoader, converterClassOrAlias, Converter.class);
+ } catch (ClassNotFoundException e) {
+ throw new ConnectException(
+ "Failed to find any class that implements Converter and which name matches "
+ + converterClassOrAlias + ", available converters are: "
+ + pluginNames(scanResult.converters())
+ );
+ }
+ break;
+ }
+ if (klass == null) {
+ throw new ConnectException("Unable to initialize the Converter specified in '" + classPropertyName + "'");
+ }
+
// Determine whether this is a key or value converter based upon the supplied property name ...
final boolean isKeyConverter = WorkerConfig.KEY_CONVERTER_CLASS_CONFIG.equals(classPropertyName);
// Configure the Converter using only the old configuration mechanism ...
String configPrefix = classPropertyName + ".";
Map<String, Object> converterConfig = config.originalsWithPrefix(configPrefix);
-
log.debug("Configuring the {} converter with configuration keys:{}{}",
- isKeyConverter ? "key" : "value", System.lineSeparator(), converterConfig.keySet());
+ isKeyConverter ? "key" : "value", System.lineSeparator(), converterConfig.keySet());
- Converter plugin = newVersionedPlugin(config, classPropertyName, versionPropertyName,
- Converter.class, classLoaderUsage, scanResult.converters());
- try (LoaderSwap loaderSwap = safeLoaderSwapper().apply(plugin.getClass().getClassLoader())) {
+ Converter plugin;
+ try (LoaderSwap loaderSwap = withClassLoader(klass.getClassLoader())) {
+ plugin = newPlugin(klass);
plugin.configure(converterConfig, isKeyConverter);
}
return plugin;
}
-
-
/**
* Load an internal converter, used by the worker for (de)serializing data in internal topics.
*
@@ -520,124 +427,99 @@ public class Plugins {
* If the given configuration defines a {@link HeaderConverter} using the named configuration property, return a new configured
* instance.
*
- * @param config the configuration containing the {@link HeaderConverter}'s configuration; may not be null
- * @param classPropertyName the name of the property that contains the name of the {@link HeaderConverter} class; may not be null
- * @param classLoaderUsage the name of the property that contains the version of the {@link HeaderConverter} class; may not be null
+ * @param config the configuration containing the {@link Converter}'s configuration; may not be null
+ * @param classPropertyName the name of the property that contains the name of the {@link Converter} class; may not be null
+ * @param classLoaderUsage which classloader should be used
* @return the instantiated and configured {@link HeaderConverter}; null if the configuration did not define the specified property
* @throws ConnectException if the {@link HeaderConverter} implementation class could not be found
*/
public HeaderConverter newHeaderConverter(AbstractConfig config, String classPropertyName, ClassLoaderUsage classLoaderUsage) {
- return newHeaderConverter(config, classPropertyName, null, classLoaderUsage);
- }
-
- /**
- * If the given configuration defines a {@link HeaderConverter} using the named configuration property, return a new configured
- * instance. If the version is specified, it will always use the plugins classloader.
- *
- * @param config the configuration containing the {@link HeaderConverter}'s configuration; may not be null
- * @param classPropertyName the name of the property that contains the name of the {@link HeaderConverter} class; may not be null
- * @param versionPropertyName the config for the version for the header converter
- * @return the instantiated and configured {@link HeaderConverter}; null if the configuration did not define the specified property
- * @throws ConnectException if the {@link HeaderConverter} implementation class could not be found
- */
- public HeaderConverter newHeaderConverter(AbstractConfig config, String classPropertyName, String versionPropertyName) {
- ClassLoaderUsage classLoader = config.getString(versionPropertyName) == null ? ClassLoaderUsage.CURRENT_CLASSLOADER : ClassLoaderUsage.PLUGINS;
- return newHeaderConverter(config, classPropertyName, versionPropertyName, classLoader);
- }
-
- private HeaderConverter newHeaderConverter(AbstractConfig config, String classPropertyName, String versionPropertyName, ClassLoaderUsage classLoaderUsage) {
- if (!config.originals().containsKey(classPropertyName) && classLoaderUsage == ClassLoaderUsage.CURRENT_CLASSLOADER) {
- // This configuration does not define the Header Converter via the specified property name
- return null;
+ Class<? extends HeaderConverter> klass = null;
+ switch (classLoaderUsage) {
+ case CURRENT_CLASSLOADER:
+ if (!config.originals().containsKey(classPropertyName)) {
+ // This connector configuration does not define the header converter via the specified property name
+ return null;
+ }
+ // Attempt to load first with the current classloader, and plugins as a fallback.
+ // Note: we can't use config.getConfiguredInstance because we have to remove the property prefixes
+ // before calling config(...)
+ klass = pluginClassFromConfig(config, classPropertyName, HeaderConverter.class, scanResult.headerConverters());
+ break;
+ case PLUGINS:
+ // Attempt to load with the plugin class loader, which uses the current classloader as a fallback.
+ // Note that there will always be at least a default header converter for the worker
+ String converterClassOrAlias = config.getClass(classPropertyName).getName();
+ try {
+ klass = pluginClass(
+ delegatingLoader,
+ converterClassOrAlias,
+ HeaderConverter.class
+ );
+ } catch (ClassNotFoundException e) {
+ throw new ConnectException(
+ "Failed to find any class that implements HeaderConverter and which name matches "
+ + converterClassOrAlias
+ + ", available header converters are: "
+ + pluginNames(scanResult.headerConverters())
+ );
+ }
+ }
+ if (klass == null) {
+ throw new ConnectException("Unable to initialize the HeaderConverter specified in '" + classPropertyName + "'");
}
- HeaderConverter plugin = newVersionedPlugin(config, classPropertyName, versionPropertyName,
- HeaderConverter.class, classLoaderUsage, scanResult.headerConverters());
String configPrefix = classPropertyName + ".";
Map<String, Object> converterConfig = config.originalsWithPrefix(configPrefix);
converterConfig.put(ConverterConfig.TYPE_CONFIG, ConverterType.HEADER.getName());
log.debug("Configuring the header converter with configuration keys:{}{}", System.lineSeparator(), converterConfig.keySet());
- try (LoaderSwap loaderSwap = safeLoaderSwapper().apply(plugin.getClass().getClassLoader())) {
+ HeaderConverter plugin;
+ try (LoaderSwap loaderSwap = withClassLoader(klass.getClassLoader())) {
+ plugin = newPlugin(klass);
plugin.configure(converterConfig);
}
return plugin;
}
- @SuppressWarnings({"unchecked", "rawtypes"})
- private <U> U newVersionedPlugin(
- AbstractConfig config,
- String classPropertyName,
- String versionPropertyName,
- Class basePluginClass,
- ClassLoaderUsage classLoaderUsage,
- SortedSet<PluginDesc<U>> availablePlugins
- ) {
-
- String version = versionPropertyName == null ? null : config.getString(versionPropertyName);
- VersionRange range = null;
- if (version != null) {
- try {
- range = PluginUtils.connectorVersionRequirement(version);
- } catch (InvalidVersionSpecificationException e) {
- throw new ConnectException(String.format("Invalid version range for %s: %s", classPropertyName, version), e);
- }
+ public ConfigProvider newConfigProvider(AbstractConfig config, String providerPrefix, ClassLoaderUsage classLoaderUsage) {
+ String classPropertyName = providerPrefix + ".class";
+ Map<String, String> originalConfig = config.originalsStrings();
+ if (!originalConfig.containsKey(classPropertyName)) {
+ // This configuration does not define the config provider via the specified property name
+ return null;
}
-
- assert range == null || classLoaderUsage == ClassLoaderUsage.PLUGINS;
-
- Class<? extends U> klass = null;
- String basePluginClassName = basePluginClass.getSimpleName();
+ Class<? extends ConfigProvider> klass = null;
switch (classLoaderUsage) {
case CURRENT_CLASSLOADER:
// Attempt to load first with the current classloader, and plugins as a fallback.
- klass = pluginClassFromConfig(config, classPropertyName, basePluginClass, availablePlugins);
+ klass = pluginClassFromConfig(config, classPropertyName, ConfigProvider.class, scanResult.configProviders());
break;
case PLUGINS:
// Attempt to load with the plugin class loader, which uses the current classloader as a fallback
-
- // if the config specifies the class name, use it, otherwise use the default which we can get from config.getClass
- String classOrAlias = config.originalsStrings().get(classPropertyName);
- if (classOrAlias == null) {
- classOrAlias = config.getClass(classPropertyName).getName();
- }
+ String configProviderClassOrAlias = originalConfig.get(classPropertyName);
try {
- klass = pluginClass(delegatingLoader, classOrAlias, basePluginClass, range);
+ klass = pluginClass(delegatingLoader, configProviderClassOrAlias, ConfigProvider.class);
} catch (ClassNotFoundException e) {
throw new ConnectException(
- "Failed to find any class that implements " + basePluginClassName + " and which name matches "
- + classOrAlias + ", available plugins are: "
- + pluginNames(availablePlugins)
+ "Failed to find any class that implements ConfigProvider and which name matches "
+ + configProviderClassOrAlias + ", available ConfigProviders are: "
+ + pluginNames(scanResult.configProviders())
);
}
break;
}
if (klass == null) {
- throw new ConnectException("Unable to initialize the " + basePluginClassName
- + " specified in " + classPropertyName);
+ throw new ConnectException("Unable to initialize the ConfigProvider specified in '" + classPropertyName + "'");
}
- U plugin;
- try (LoaderSwap loaderSwap = withClassLoader(klass.getClassLoader())) {
- plugin = newPlugin(klass);
- }
- return plugin;
- }
-
- public ConfigProvider newConfigProvider(AbstractConfig config, String providerPrefix, ClassLoaderUsage classLoaderUsage) {
- String classPropertyName = providerPrefix + ".class";
- Map<String, String> originalConfig = config.originalsStrings();
- if (!originalConfig.containsKey(classPropertyName)) {
- // This configuration does not define the config provider via the specified property name
- return null;
- }
-
- ConfigProvider plugin = newVersionedPlugin(config, classPropertyName, null, ConfigProvider.class, classLoaderUsage, scanResult.configProviders());
-
// Configure the ConfigProvider
String configPrefix = providerPrefix + ".param.";
Map<String, Object> configProviderConfig = config.originalsWithPrefix(configPrefix);
- try (LoaderSwap loaderSwap = safeLoaderSwapper().apply(plugin.getClass().getClassLoader())) {
+
+ ConfigProvider plugin;
+ try (LoaderSwap loaderSwap = withClassLoader(klass.getClassLoader())) {
+ plugin = newPlugin(klass);
plugin.configure(configProviderConfig);
}
return plugin;
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginsRecommenders.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginsRecommenders.java
deleted file mode 100644
index 8cf209ac08c..00000000000
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginsRecommenders.java
+++ /dev/null
@@ -1,198 +0,0 @@
-/*
- * 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.isolation;
-
-import org.apache.kafka.common.config.ConfigDef;
-import org.apache.kafka.connect.runtime.ConnectorConfig;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
-public class PluginsRecommenders {
-
- private final Plugins plugins;
- private final ConverterPluginRecommender converterPluginRecommender;
- private final ConnectorPluginVersionRecommender connectorPluginVersionRecommender;
- private final HeaderConverterPluginRecommender headerConverterPluginRecommender;
- private final KeyConverterPluginVersionRecommender keyConverterPluginVersionRecommender;
- private final ValueConverterPluginVersionRecommender valueConverterPluginVersionRecommender;
- private final HeaderConverterPluginVersionRecommender headerConverterPluginVersionRecommender;
-
- public PluginsRecommenders() {
- this(null);
- }
-
- public PluginsRecommenders(Plugins plugins) {
- this.plugins = plugins;
- this.converterPluginRecommender = new ConverterPluginRecommender();
- this.connectorPluginVersionRecommender = new ConnectorPluginVersionRecommender();
- this.headerConverterPluginRecommender = new HeaderConverterPluginRecommender();
- this.keyConverterPluginVersionRecommender = new KeyConverterPluginVersionRecommender();
- this.valueConverterPluginVersionRecommender = new ValueConverterPluginVersionRecommender();
- this.headerConverterPluginVersionRecommender = new HeaderConverterPluginVersionRecommender();
- }
-
- public ConverterPluginRecommender converterPluginRecommender() {
- return converterPluginRecommender;
- }
-
- public ConnectorPluginVersionRecommender connectorPluginVersionRecommender() {
- return connectorPluginVersionRecommender;
- }
-
- public HeaderConverterPluginRecommender headerConverterPluginRecommender() {
- return headerConverterPluginRecommender;
- }
-
- public KeyConverterPluginVersionRecommender keyConverterPluginVersionRecommender() {
- return keyConverterPluginVersionRecommender;
- }
-
- public ValueConverterPluginVersionRecommender valueConverterPluginVersionRecommender() {
- return valueConverterPluginVersionRecommender;
- }
-
- public HeaderConverterPluginVersionRecommender headerConverterPluginVersionRecommender() {
- return headerConverterPluginVersionRecommender;
- }
-
- public class ConnectorPluginVersionRecommender implements ConfigDef.Recommender {
-
- @SuppressWarnings({"unchecked", "rawtypes"})
- @Override
- public List<Object> validValues(String name, Map<String, Object> parsedConfig) {
- if (plugins == null) {
- return Collections.emptyList();
- }
- String connectorClassOrAlias = (String) parsedConfig.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
- if (connectorClassOrAlias == null) {
- //should never happen
- return Collections.emptyList();
- }
- List<Object> sourceConnectors = plugins.sourceConnectors(connectorClassOrAlias).stream()
- .map(PluginDesc::version).distinct().collect(Collectors.toList());
- if (!sourceConnectors.isEmpty()) {
- return sourceConnectors;
- }
- return plugins.sinkConnectors(connectorClassOrAlias).stream()
- .map(PluginDesc::version).distinct().collect(Collectors.toList());
- }
-
- @Override
- public boolean visible(String name, Map<String, Object> parsedConfig) {
- return parsedConfig.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG) != null;
- }
-
- }
-
- public class ConverterPluginRecommender implements ConfigDef.Recommender {
-
- @Override
- public List<Object> validValues(String name, Map<String, Object> parsedConfig) {
- if (plugins == null) {
- return Collections.emptyList();
- }
- return plugins.converters().stream()
- .map(PluginDesc::pluginClass).distinct().collect(Collectors.toList());
- }
-
- @Override
- public boolean visible(String name, Map<String, Object> parsedConfig) {
- return true;
- }
- }
-
- public class HeaderConverterPluginRecommender implements ConfigDef.Recommender {
-
- @Override
- public List<Object> validValues(String name, Map<String, Object> parsedConfig) {
- if (plugins == null) {
- return Collections.emptyList();
- }
- return plugins.headerConverters().stream()
- .map(PluginDesc::pluginClass).distinct().collect(Collectors.toList());
- }
-
- @Override
- public boolean visible(String name, Map<String, Object> parsedConfig) {
- return true;
- }
- }
-
- public abstract class ConverterPluginVersionRecommender implements ConfigDef.Recommender {
-
- protected Function<String, List<Object>> recommendations() {
- return converterClass -> plugins.converters(converterClass).stream()
- .map(PluginDesc::version).distinct().collect(Collectors.toList());
- }
-
- protected abstract String converterConfig();
-
- @SuppressWarnings({"rawtypes"})
- @Override
- public List<Object> validValues(String name, Map<String, Object> parsedConfig) {
- if (plugins == null) {
- return Collections.emptyList();
- }
- if (parsedConfig.get(converterConfig()) == null) {
- return Collections.emptyList();
- }
- Class converterClass = (Class) parsedConfig.get(converterConfig());
- return recommendations().apply(converterClass.getName());
- }
-
- @Override
- public boolean visible(String name, Map<String, Object> parsedConfig) {
- return parsedConfig.get(converterConfig()) != null;
- }
- }
-
- public class KeyConverterPluginVersionRecommender extends ConverterPluginVersionRecommender {
-
- @Override
- protected String converterConfig() {
- return ConnectorConfig.KEY_CONVERTER_CLASS_CONFIG;
- }
-
- }
-
- public class ValueConverterPluginVersionRecommender extends ConverterPluginVersionRecommender {
-
- @Override
- protected String converterConfig() {
- return ConnectorConfig.VALUE_CONVERTER_CLASS_CONFIG;
- }
- }
-
- public class HeaderConverterPluginVersionRecommender extends ConverterPluginVersionRecommender {
-
- @Override
- protected String converterConfig() {
- return ConnectorConfig.HEADER_CONVERTER_CLASS_CONFIG;
- }
-
- @Override
- protected Function<String, List<Object>> recommendations() {
- return converterClass -> plugins.headerConverters(converterClass).stream()
- .map(PluginDesc::version).distinct().collect(Collectors.toList());
- }
- }
-}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/VersionedPluginLoadingException.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/VersionedPluginLoadingException.java
deleted file mode 100644
index 8fb3042549b..00000000000
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/VersionedPluginLoadingException.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * 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.isolation;
-
-import org.apache.kafka.common.config.ConfigException;
-
-import java.util.List;
-
-public class VersionedPluginLoadingException extends ConfigException {
-
- private List<String> availableVersions = null;
-
- public VersionedPluginLoadingException(String message) {
- super(message);
- }
-
- public VersionedPluginLoadingException(String message, List<String> availableVersions) {
- super(message);
- this.availableVersions = availableVersions;
- }
-
- public List<String> availableVersions() {
- return availableVersions;
- }
-}
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 46c78027f25..800a8b2c1a3 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
@@ -20,7 +20,6 @@ import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.Herder;
import org.apache.kafka.connect.runtime.isolation.PluginDesc;
import org.apache.kafka.connect.runtime.isolation.PluginType;
-import org.apache.kafka.connect.runtime.isolation.PluginUtils;
import org.apache.kafka.connect.runtime.rest.RestRequestTimeout;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos;
import org.apache.kafka.connect.runtime.rest.entities.ConfigKeyInfo;
@@ -30,9 +29,6 @@ import org.apache.kafka.connect.util.FutureCallback;
import org.apache.kafka.connect.util.Stage;
import org.apache.kafka.connect.util.StagedTimeoutException;
-import org.apache.maven.artifact.versioning.InvalidVersionSpecificationException;
-import org.apache.maven.artifact.versioning.VersionRange;
-
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection;
@@ -157,18 +153,9 @@ public class ConnectorPluginsResource {
@GET
@Path("/{pluginName}/config")
@Operation(summary = "Get the configuration definition for the specified pluginName")
- public List<ConfigKeyInfo> getConnectorConfigDef(final @PathParam("pluginName") String pluginName,
- final @QueryParam("version") @DefaultValue("latest") String version) {
-
- VersionRange range = null;
- try {
- range = PluginUtils.connectorVersionRequirement(version);
- } catch (InvalidVersionSpecificationException e) {
- throw new BadRequestException("Invalid version specification: " + version, e);
- }
-
+ public List<ConfigKeyInfo> getConnectorConfigDef(final @PathParam("pluginName") String pluginName) {
synchronized (this) {
- return herder.connectorPluginConfig(pluginName, range);
+ return herder.connectorPluginConfig(pluginName);
}
}
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 01a9bcf6373..aa715667d24 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
@@ -50,7 +50,6 @@ import org.apache.kafka.connect.runtime.rest.errors.BadRequestException;
import org.apache.kafka.connect.storage.AppliedConnectorConfig;
import org.apache.kafka.connect.storage.ClusterConfigState;
import org.apache.kafka.connect.storage.ConfigBackingStore;
-import org.apache.kafka.connect.storage.SimpleHeaderConverter;
import org.apache.kafka.connect.storage.StatusBackingStore;
import org.apache.kafka.connect.transforms.Transformation;
import org.apache.kafka.connect.transforms.predicates.Predicate;
@@ -178,7 +177,6 @@ public class AbstractHerderTest {
private final ConnectorClientConfigOverridePolicy noneConnectorClientConfigOverridePolicy = new NoneConnectorClientConfigOverridePolicy();
@Mock private Worker worker;
- @Mock private WorkerConfig workerConfig;
@Mock private WorkerConfigTransformer transformer;
@Mock private ConfigBackingStore configStore;
@Mock private StatusBackingStore statusStore;
@@ -208,11 +206,11 @@ public class AbstractHerderTest {
public void testConnectorStatus() {
ConnectorTaskId taskId = new ConnectorTaskId(connectorName, 0);
- when(plugins.newConnector(anyString(), any())).thenReturn(new SampleSourceConnector());
- when(worker.getPlugins()).thenReturn(plugins);
-
AbstractHerder herder = testHerder();
+ when(plugins.newConnector(anyString())).thenReturn(new SampleSourceConnector());
+ when(herder.plugins()).thenReturn(plugins);
+
when(herder.rawConfig(connectorName)).thenReturn(Collections.singletonMap(
ConnectorConfig.CONNECTOR_CLASS_CONFIG, SampleSourceConnector.class.getName()
));
@@ -242,11 +240,11 @@ public class AbstractHerderTest {
public void testConnectorStatusMissingPlugin() {
ConnectorTaskId taskId = new ConnectorTaskId(connectorName, 0);
- when(plugins.newConnector(anyString(), any())).thenThrow(new ConnectException("Unable to find class"));
- when(worker.getPlugins()).thenReturn(plugins);
-
AbstractHerder herder = testHerder();
+ when(plugins.newConnector(anyString())).thenThrow(new ConnectException("Unable to find class"));
+ when(herder.plugins()).thenReturn(plugins);
+
when(herder.rawConfig(connectorName))
.thenReturn(Collections.singletonMap(ConnectorConfig.CONNECTOR_CLASS_CONFIG, "missing"));
@@ -273,12 +271,11 @@ public class AbstractHerderTest {
@Test
public void testConnectorInfo() {
-
- when(plugins.newConnector(anyString(), any())).thenReturn(new SampleSourceConnector());
- when(worker.getPlugins()).thenReturn(plugins);
-
AbstractHerder herder = testHerder();
+ when(plugins.newConnector(anyString())).thenReturn(new SampleSourceConnector());
+ when(herder.plugins()).thenReturn(plugins);
+
when(configStore.snapshot()).thenReturn(SNAPSHOT);
ConnectorInfo info = herder.connectorInfo(CONN1);
@@ -313,12 +310,11 @@ public class AbstractHerderTest {
@Test
public void testConnectorInfoMissingPlugin() {
-
- when(plugins.newConnector(anyString(), any())).thenThrow(new ConnectException("No class found"));
- when(worker.getPlugins()).thenReturn(plugins);
-
AbstractHerder herder = testHerder();
+ when(plugins.newConnector(anyString())).thenThrow(new ConnectException("No class found"));
+ when(herder.plugins()).thenReturn(plugins);
+
when(configStore.snapshot()).thenReturn(SNAPSHOT);
ConnectorInfo info = herder.connectorInfo(CONN1);
@@ -485,7 +481,7 @@ public class AbstractHerderTest {
Map<String, ConfigInfo> infos = result.values().stream()
.collect(Collectors.toMap(info -> info.configKey().name(), Function.identity()));
// Base connector config has 15 fields, connector's configs add 7
- assertEquals(26, infos.size());
+ assertEquals(22, infos.size());
// Missing name should generate an error
assertEquals(ConnectorConfig.NAME_CONFIG,
infos.get(ConnectorConfig.NAME_CONFIG).configValue().name());
@@ -596,7 +592,7 @@ public class AbstractHerderTest {
assertEquals(1, result.errorCount());
Map<String, ConfigInfo> infos = result.values().stream()
.collect(Collectors.toMap(info -> info.configKey().name(), Function.identity()));
- assertEquals(31, infos.size());
+ assertEquals(27, infos.size());
// Should get 2 type fields from the transforms, first adds its own config since it has a valid class
assertEquals("transforms.xformA.type",
infos.get("transforms.xformA.type").configValue().name());
@@ -653,7 +649,7 @@ public class AbstractHerderTest {
assertEquals(1, result.errorCount());
Map<String, ConfigInfo> infos = result.values().stream()
.collect(Collectors.toMap(info -> info.configKey().name(), Function.identity()));
- assertEquals(33, infos.size());
+ assertEquals(29, infos.size());
// Should get 2 type fields from the transforms, first adds its own config since it has a valid class
assertEquals("transforms.xformA.type", infos.get("transforms.xformA.type").configValue().name());
assertTrue(infos.get("transforms.xformA.type").configValue().errors().isEmpty());
@@ -714,8 +710,8 @@ public class AbstractHerderTest {
);
assertEquals(expectedGroups, result.groups());
assertEquals(1, result.errorCount());
- // Base connector config has 19 fields, connector's configs add 7, and 2 producer overrides
- assertEquals(28, result.values().size());
+ // Base connector config has 15 fields, connector's configs add 7, and 2 producer overrides
+ assertEquals(24, result.values().size());
assertTrue(result.values().stream().anyMatch(
configInfo -> ackConfigKey.equals(configInfo.configValue().name()) && !configInfo.configValue().errors().isEmpty()));
assertTrue(result.values().stream().anyMatch(
@@ -1044,8 +1040,8 @@ public class AbstractHerderTest {
) throws ClassNotFoundException {
AbstractHerder herder = testHerder();
- when(plugins.pluginClass(pluginName, null)).then(invocation -> newPluginInstance.get().getClass());
- when(plugins.newPlugin(anyString(), any())).then(invocation -> newPluginInstance.get());
+ when(plugins.pluginClass(pluginName)).then(invocation -> newPluginInstance.get().getClass());
+ when(plugins.newPlugin(anyString())).then(invocation -> newPluginInstance.get());
when(herder.plugins()).thenReturn(plugins);
List<ConfigKeyInfo> configs = herder.connectorPluginConfig(pluginName);
@@ -1064,7 +1060,7 @@ public class AbstractHerderTest {
String connName = "AnotherPlugin";
AbstractHerder herder = testHerder();
when(worker.getPlugins()).thenReturn(plugins);
- when(plugins.pluginClass(anyString(), any())).thenThrow(new ClassNotFoundException());
+ when(plugins.pluginClass(anyString())).thenThrow(new ClassNotFoundException());
assertThrows(NotFoundException.class, () -> herder.connectorPluginConfig(connName));
}
@@ -1074,17 +1070,17 @@ public class AbstractHerderTest {
String connName = "AnotherPlugin";
AbstractHerder herder = testHerder();
when(worker.getPlugins()).thenReturn(plugins);
- when(plugins.pluginClass(anyString(), any())).thenReturn((Class) Object.class);
- when(plugins.newPlugin(anyString(), any())).thenReturn(new DirectoryConfigProvider());
+ when(plugins.pluginClass(anyString())).thenReturn((Class) Object.class);
+ when(plugins.newPlugin(anyString())).thenReturn(new DirectoryConfigProvider());
assertThrows(BadRequestException.class, () -> herder.connectorPluginConfig(connName));
}
@Test
public void testGetConnectorTypeWithMissingPlugin() {
String connName = "AnotherPlugin";
- when(worker.getPlugins()).thenReturn(plugins);
- when(plugins.newConnector(anyString(), any())).thenThrow(new ConnectException("No class found"));
AbstractHerder herder = testHerder();
+ when(worker.getPlugins()).thenReturn(plugins);
+ when(plugins.newConnector(anyString())).thenThrow(new ConnectException("No class found"));
assertEquals(ConnectorType.UNKNOWN, herder.connectorType(Collections.singletonMap(ConnectorConfig.CONNECTOR_CLASS_CONFIG, connName)));
}
@@ -1248,14 +1244,15 @@ public class AbstractHerderTest {
private AbstractHerder createConfigValidationHerder(Class<? extends Connector> connectorClass,
ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy,
int countOfCallingNewConnector) {
+
+ AbstractHerder herder = testHerder(connectorClientConfigOverridePolicy);
+
// Call to validateConnectorConfig
when(worker.configTransformer()).thenReturn(transformer);
@SuppressWarnings("unchecked")
final ArgumentCaptor<Map<String, String>> mapArgumentCaptor = ArgumentCaptor.forClass(Map.class);
when(transformer.transform(mapArgumentCaptor.capture())).thenAnswer(invocation -> mapArgumentCaptor.getValue());
when(worker.getPlugins()).thenReturn(plugins);
-
- AbstractHerder herder = testHerder(connectorClientConfigOverridePolicy);
final Connector connector;
try {
connector = connectorClass.getConstructor().newInstance();
@@ -1278,17 +1275,14 @@ public class AbstractHerderTest {
.defaultAnswer(CALLS_REAL_METHODS));
}
- @SuppressWarnings({"unchecked", "rawtypes"})
private void mockValidationIsolation(String connectorClass, Connector connector) {
- when(workerConfig.getClass(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG)).thenReturn((Class) SimpleHeaderConverter.class);
- when(worker.config()).thenReturn(workerConfig);
- when(plugins.newConnector(anyString(), any())).thenReturn(connector);
- when(plugins.pluginLoader(connectorClass, null)).thenReturn(classLoader);
+ when(plugins.newConnector(connectorClass)).thenReturn(connector);
+ when(plugins.connectorLoader(connectorClass)).thenReturn(classLoader);
when(plugins.withClassLoader(classLoader)).thenReturn(loaderSwap);
}
private void verifyValidationIsolation() {
- verify(plugins).newConnector(anyString(), any());
+ verify(plugins).newConnector(anyString());
verify(plugins).withClassLoader(classLoader);
verify(loaderSwap).close();
}
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleSourceConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleSourceConnector.java
index 1324b9a2263..cb91530439f 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleSourceConnector.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleSourceConnector.java
@@ -27,7 +27,7 @@ import java.util.Map;
public class SampleSourceConnector extends SourceConnector {
- public static final String VERSION = "some great version";
+ public static final String VERSION = "an entirely different version";
@Override
public String version() {
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java
index 74da1703482..2632360b7f6 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java
@@ -320,7 +320,7 @@ public class DistributedHerderTest {
herder = mock(DistributedHerder.class, withSettings().defaultAnswer(CALLS_REAL_METHODS).useConstructor(new DistributedConfig(HERDER_CONFIG),
worker, WORKER_ID, KAFKA_CLUSTER_ID, statusBackingStore, configBackingStore, member, MEMBER_URL, restClient, metrics, time,
noneConnectorClientConfigOverridePolicy, Collections.emptyList(), null, new AutoCloseable[]{uponShutdown}));
- verify(worker).getPlugins();
+
configUpdateListener = herder.new ConfigUpdateListener();
rebalanceListener = herder.new RebalanceListener(time);
conn1SinkConfig = new SinkConnectorConfig(plugins, CONN1_CONFIG);
@@ -3550,7 +3550,7 @@ public class DistributedHerderTest {
herder = mock(DistributedHerder.class, withSettings().defaultAnswer(CALLS_REAL_METHODS).useConstructor(new DistributedConfig(HERDER_CONFIG),
worker, WORKER_ID, KAFKA_CLUSTER_ID, statusBackingStore, configBackingStore, member, MEMBER_URL, restClient, metrics, time,
noneConnectorClientConfigOverridePolicy, Collections.emptyList(), new MockSynchronousExecutor(), new AutoCloseable[]{}));
- verify(worker, times(2)).getPlugins();
+
rebalanceListener = herder.new RebalanceListener(time);
when(member.memberId()).thenReturn("member");
@@ -3992,7 +3992,6 @@ public class DistributedHerderTest {
public void testModifyOffsetsSourceConnectorExactlyOnceEnabled() throws Exception {
// Setup herder with exactly-once support for source connectors enabled
herder = exactlyOnceHerder();
- verify(worker, times(2)).getPlugins();
rebalanceListener = herder.new RebalanceListener(time);
// Get the initial assignment
when(member.memberId()).thenReturn("leader");
@@ -4058,7 +4057,6 @@ public class DistributedHerderTest {
public void testModifyOffsetsSourceConnectorExactlyOnceEnabledZombieFencingFailure() {
// Setup herder with exactly-once support for source connectors enabled
herder = exactlyOnceHerder();
- verify(worker, times(2)).getPlugins();
rebalanceListener = herder.new RebalanceListener(time);
// Get the initial assignment
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/SynchronizationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/SynchronizationTest.java
index d0559123b72..04501e07638 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/SynchronizationTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/SynchronizationTest.java
@@ -27,7 +27,6 @@ import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.connect.runtime.WorkerConfig;
import org.apache.kafka.connect.storage.Converter;
-import org.apache.maven.artifact.versioning.VersionRange;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -191,10 +190,10 @@ public class SynchronizationTest {
}
@Override
- public PluginClassLoader pluginClassLoader(String name, VersionRange range) {
+ public PluginClassLoader pluginClassLoader(String name) {
dclBreakpoint.await(name);
dclBreakpoint.await(name);
- return super.pluginClassLoader(name, range);
+ return super.pluginClassLoader(name);
}
}
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 1b8376db635..bec99d2d55c 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
@@ -428,14 +428,14 @@ public class ConnectorPluginsResourceTest {
@Test
public void testGetConnectorConfigDef() {
String connName = ConnectorPluginsResourceTestConnector.class.getName();
- when(herder.connectorPluginConfig(eq(connName), eq(null))).thenAnswer(answer -> {
+ 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, null);
+ 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();
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java
index 17a7d7c391a..c220ca9c70d 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java
@@ -38,7 +38,6 @@ import org.apache.kafka.connect.runtime.TargetState;
import org.apache.kafka.connect.runtime.TaskConfig;
import org.apache.kafka.connect.runtime.TaskStatus;
import org.apache.kafka.connect.runtime.Worker;
-import org.apache.kafka.connect.runtime.WorkerConfig;
import org.apache.kafka.connect.runtime.WorkerConfigTransformer;
import org.apache.kafka.connect.runtime.distributed.SampleConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.runtime.isolation.LoaderSwap;
@@ -57,7 +56,6 @@ import org.apache.kafka.connect.source.SourceTask;
import org.apache.kafka.connect.storage.AppliedConnectorConfig;
import org.apache.kafka.connect.storage.ClusterConfigState;
import org.apache.kafka.connect.storage.MemoryConfigBackingStore;
-import org.apache.kafka.connect.storage.SimpleHeaderConverter;
import org.apache.kafka.connect.storage.StatusBackingStore;
import org.apache.kafka.connect.util.Callback;
import org.apache.kafka.connect.util.ConnectorTaskId;
@@ -132,8 +130,6 @@ public class StandaloneHerderTest {
@Mock
protected Worker worker;
@Mock
- protected WorkerConfig workerConfig;
- @Mock
protected WorkerConfigTransformer transformer;
@Mock
private Plugins plugins;
@@ -148,11 +144,9 @@ public class StandaloneHerderTest {
noneConnectorClientConfigOverridePolicy = new SampleConnectorClientConfigOverridePolicy();
public void initialize(boolean mockTransform) {
- when(worker.getPlugins()).thenReturn(plugins);
herder = mock(StandaloneHerder.class, withSettings()
.useConstructor(worker, WORKER_ID, KAFKA_CLUSTER_ID, statusBackingStore, new MemoryConfigBackingStore(transformer), noneConnectorClientConfigOverridePolicy, new MockTime())
.defaultAnswer(CALLS_REAL_METHODS));
- verify(worker).getPlugins();
createCallback = new FutureCallback<>();
final ArgumentCaptor<Map<String, String>> configCapture = ArgumentCaptor.forClass(Map.class);
if (mockTransform)
@@ -179,7 +173,6 @@ public class StandaloneHerderTest {
}
@Test
- @SuppressWarnings("rawtypes")
public void testCreateConnectorFailedValidation() {
initialize(false);
// Basic validation should be performed and return an error, but should still evaluate the connector's config
@@ -192,13 +185,12 @@ public class StandaloneHerderTest {
final ArgumentCaptor<Map<String, String>> configCapture = ArgumentCaptor.forClass(Map.class);
when(transformer.transform(configCapture.capture())).thenAnswer(invocation -> configCapture.getValue());
when(worker.getPlugins()).thenReturn(plugins);
- when(worker.config()).thenReturn(workerConfig);
- when(workerConfig.getClass(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG)).thenReturn((Class) SimpleHeaderConverter.class);
- when(plugins.newConnector(anyString(), any())).thenReturn(connectorMock);
- when(plugins.pluginLoader(anyString(), any())).thenReturn(pluginLoader);
+ when(plugins.newConnector(anyString())).thenReturn(connectorMock);
+ when(plugins.connectorLoader(anyString())).thenReturn(pluginLoader);
when(plugins.withClassLoader(pluginLoader)).thenReturn(loaderSwap);
when(connectorMock.config()).thenReturn(new ConfigDef());
+
ConfigValue validatedValue = new ConfigValue("foo.bar");
when(connectorMock.validate(config)).thenReturn(new Config(singletonList(validatedValue)));
@@ -858,7 +850,6 @@ public class StandaloneHerderTest {
}
@Test
- @SuppressWarnings("rawtypes")
public void testCorruptConfig() {
initialize(false);
Map<String, String> config = new HashMap<>();
@@ -879,12 +870,10 @@ public class StandaloneHerderTest {
when(worker.configTransformer()).thenReturn(transformer);
final ArgumentCaptor<Map<String, String>> configCapture = ArgumentCaptor.forClass(Map.class);
when(transformer.transform(configCapture.capture())).thenAnswer(invocation -> configCapture.getValue());
- when(worker.config()).thenReturn(workerConfig);
when(worker.getPlugins()).thenReturn(plugins);
- when(workerConfig.getClass(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG)).thenReturn((Class) SimpleHeaderConverter.class);
- when(plugins.pluginLoader(anyString(), any())).thenReturn(pluginLoader);
+ when(plugins.connectorLoader(anyString())).thenReturn(pluginLoader);
when(plugins.withClassLoader(pluginLoader)).thenReturn(loaderSwap);
- when(plugins.newConnector(anyString(), any())).thenReturn(connectorMock);
+ when(plugins.newConnector(anyString())).thenReturn(connectorMock);
when(connectorMock.config()).thenReturn(configDef);
herder.putConnectorConfig(CONNECTOR_NAME, config, true, createCallback);
@@ -1223,7 +1212,6 @@ public class StandaloneHerderTest {
return generatedTaskProps;
}
- @SuppressWarnings("rawtypes")
private void expectConfigValidation(
SourceSink sourceSink,
Map<String, String>... configs
@@ -1233,13 +1221,13 @@ public class StandaloneHerderTest {
when(worker.configTransformer()).thenReturn(transformer);
final ArgumentCaptor<Map<String, String>> configCapture = ArgumentCaptor.forClass(Map.class);
when(transformer.transform(configCapture.capture())).thenAnswer(invocation -> configCapture.getValue());
- when(worker.config()).thenReturn(workerConfig);
- when(workerConfig.getClass(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG)).thenReturn((Class) SimpleHeaderConverter.class);
- when(plugins.pluginLoader(anyString(), any())).thenReturn(pluginLoader);
+ when(worker.getPlugins()).thenReturn(plugins);
+ when(plugins.connectorLoader(anyString())).thenReturn(pluginLoader);
when(plugins.withClassLoader(pluginLoader)).thenReturn(loaderSwap);
+
// Assume the connector should always be created
when(worker.getPlugins()).thenReturn(plugins);
- when(plugins.newConnector(anyString(), any())).thenReturn(connectorMock);
+ when(plugins.newConnector(anyString())).thenReturn(connectorMock);
when(connectorMock.config()).thenReturn(new ConfigDef());
// Set up validation for each config