Skip to content

KAFKA-13327, KAFKA-13328, KAFKA-13329: Clean up preflight connector validation #11369

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
C0urante wants to merge 5 commits into from
Closed

KAFKA-13327, KAFKA-13328, KAFKA-13329: Clean up preflight connector validation #11369

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
68cbb72
KAFKA-13327, KAFKA-13328, KAFKA-13329: Clean up preflight connector v…
C0urante Sep 29, 2021
7cd6b74
Address Checkstyle grievances
C0urante Sep 30, 2021
9f7a279
Update connect/runtime/src/main/java/org/apache/kafka/connect/runtime…
C0urante Oct 1, 2021
03252c1
Address review comments and fix sink connector config validation logic
C0urante Oct 1, 2021
fa9531e
Update AbstractHerder.java
C0urante Oct 28, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,7 @@
files="(KafkaConfigBackingStore|Values).java"/>

<suppress checks="NPathComplexity"
files="(DistributedHerder|RestClient|RestServer|JsonConverter|KafkaConfigBackingStore|FileStreamSourceTask|TopicAdmin).java"/>
files="(AbstractHerder|DistributedHerder|RestClient|RestServer|JsonConverter|KafkaConfigBackingStore|FileStreamSourceTask|TopicAdmin).java"/>

<!-- connect tests-->
<suppress checks="ClassDataAbstractionCoupling"
Expand Down
72 changes: 72 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,11 @@
*/
package org.apache.kafka.common.config;

import java.lang.reflect.Modifier;
import java.util.function.Function;
import java.util.stream.Collectors;

import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.types.Password;
import org.apache.kafka.common.utils.Utils;

Expand All @@ -35,6 +38,7 @@
import java.util.function.BiConsumer;
import java.util.function.Supplier;
import java.util.regex.Pattern;
import java.util.stream.Stream;

/**
* This class is used for specifying the set of expected configurations. For each configuration, you can specify
Expand Down Expand Up @@ -1116,11 +1120,79 @@ public void ensureValid(String name, Object value) {
}
}

@Override
public String toString() {
return "non-empty string without ISO control characters";
}
}

public static class InstantiableClassValidator implements Validator {
@Override
public void ensureValid(String name, Object value) {
if (value == null) {
// The value will be null if the class couldn't be found; no point in performing follow-up validation
return;
}

Class<?> cls = (Class<?>) value;
try {
cls.getDeclaredConstructor().newInstance();

@gharris1727 gharris1727 Sep 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I hope these arbitrary classes that people put in here don't cause memory leaks from doing their own initialization.

@C0urante C0urante Oct 1, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Think it's worth adding a warning about that to the Javadoc for the class?

@gharris1727 gharris1727 Oct 1, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, I don't think that buys us much, as the developer using this validator won't have much more insight into the arbitrary classes' behavior than the developer writing this validator.
Perhaps we could combine these two validators, or make the execution of this one dependent on the subclass validator, to try to narrow down the range of classes that could be instantiated here. For example, if the user is supposed to provide a HeaderConverter class, the typical HeaderConverter class will have a non-leaky default constructor.
...Except i just looked through the API definitions, and I don't think the javadocs for any of our API classes talk about default constructors and expectations, I'd have to dig deep to find out where this expectation is documented, if it exists. Maybe checking the subclass doesn't buy ourselves anything.

If you want to write a validator like this, I think the only justification we can have is that if it causes a memory leak, that's the implementation's fault because the framework never had any guarantees on the number of objects instantiated.

LGTM.

} catch (NoSuchMethodException e) {
throw new ConfigException(name, cls.getName(), "Could not find a public no-argument constructor for class" + (e.getMessage() != null ? ": " + e.getMessage() : ""));
} catch (ReflectiveOperationException | RuntimeException e) {
throw new ConfigException(name, cls.getName(), "Could not instantiate class" + (e.getMessage() != null ? ": " + e.getMessage() : ""));
}
}

@Override
public String toString() {
return "A class with a public, no-argument constructor";
}
}

public static class ConcreteSubClassValidator implements Validator {
private final Class<?> expectedSuperClass;

private ConcreteSubClassValidator(Class<?> expectedSuperClass) {
this.expectedSuperClass = expectedSuperClass;
}

public static ConcreteSubClassValidator forSuperClass(Class<?> expectedSuperClass) {
return new ConcreteSubClassValidator(expectedSuperClass);
}

@Override
public void ensureValid(String name, Object value) {
if (value == null) {
// The value will be null if the class couldn't be found; no point in performing follow-up validation
return;
}

Class<?> cls = (Class<?>) value;
if (!expectedSuperClass.isAssignableFrom(cls)) {
throw new ConfigException(name, String.valueOf(cls), "Not a " + expectedSuperClass.getSimpleName());
}

if (Modifier.isAbstract(cls.getModifiers())) {
String childClassNames = Stream.of(cls.getClasses())
.filter(cls::isAssignableFrom)
.filter(c -> !Modifier.isAbstract(c.getModifiers()))
.filter(c -> Modifier.isPublic(c.getModifiers()))
.map(Class::getName)
.collect(Collectors.joining(", "));
String message = Utils.isBlank(childClassNames) ?
"Class is abstract and cannot be created." :
"Class is abstract and cannot be created. Did you mean " + childClassNames + "?";
throw new ConfigException(name, cls.getName(), message);
}
}

@Override
public String toString() {
return "A concrete subclass of " + expectedSuperClass.getName();
}
}

public static class ConfigKey {
public final String name;
public final Type type;
Expand Down
151 changes: 115 additions & 36 deletions connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.kafka.common.config.ConfigDef.Type;
import org.apache.kafka.common.config.ConfigTransformer;
import org.apache.kafka.common.config.ConfigValue;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.connect.connector.Connector;
import org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.connector.policy.ConnectorClientConfigRequest;
Expand All @@ -41,9 +42,14 @@
import org.apache.kafka.connect.runtime.rest.errors.BadRequestException;
import org.apache.kafka.connect.source.SourceConnector;
import org.apache.kafka.connect.storage.ConfigBackingStore;
import org.apache.kafka.connect.storage.ConverterConfig;
import org.apache.kafka.connect.storage.ConverterType;
import org.apache.kafka.connect.storage.HeaderConverter;
import org.apache.kafka.connect.storage.StatusBackingStore;
import org.apache.kafka.connect.util.Callback;
import org.apache.kafka.connect.util.ConnectorTaskId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
Expand All @@ -69,6 +75,8 @@
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static org.apache.kafka.connect.runtime.ConnectorConfig.HEADER_CONVERTER_CLASS_CONFIG;

/**
* Abstract Herder implementation which handles connector/task lifecycle tracking. Extensions
* must invoke the lifecycle hooks appropriately.
Expand All @@ -92,6 +100,8 @@
*/
public abstract class AbstractHerder implements Herder, TaskStatus.Listener, ConnectorStatus.Listener {

private static final Logger log = LoggerFactory.getLogger(AbstractHerder.class);

private final String workerId;
protected final Worker worker;
private final String kafkaClusterId;
Expand Down Expand Up @@ -344,12 +354,73 @@ public ConnectorStateInfo.TaskState taskStatus(ConnectorTaskId id) {
status.workerId(), status.trace());
}

protected Map<String, ConfigValue> validateBasicConnectorConfig(Connector connector,
ConfigDef configDef,
Map<String, String> config) {
protected Map<String, ConfigValue> validateSinkConnectorConfig(ConfigDef configDef, Map<String, String> config) {
return SinkConnectorConfig.validate(configDef.validateAll(config), config);
}

protected Map<String, ConfigValue> validateSourceConnectorConfig(ConfigDef configDef, Map<String, String> config) {
return configDef.validateAll(config);
}

private ConfigInfos validateHeaderConverterConfig(Map<String, String> connectorConfig, ConfigValue headerConverterConfigValue) {
String headerConverterClass = connectorConfig.get(HEADER_CONVERTER_CLASS_CONFIG);

if (headerConverterClass == null
|| headerConverterConfigValue == null
|| !headerConverterConfigValue.errorMessages().isEmpty()
) {
// Either no custom header converter was specified, or one was specified but there's a problem with it.
// No need to proceed any further.
return null;
}
Comment on lines +368 to +375

@C0urante C0urante Sep 29, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The short-circuiting pattern here where we conditionally exit the method and otherwise proceed normally doesn't seem to be playing very nicely with Checkstyle's NPath complexity metric. I kept this style because it reduces indentation and seems more readable than the alternative (where the portions of code after an if statement such as this one would be put inside an else block), but it comes at the cost that we have to add a Checkstyle suppression.


HeaderConverter headerConverter;
try {
headerConverter = Utils.newInstance(headerConverterClass, HeaderConverter.class);
} catch (ClassNotFoundException | RuntimeException e) {
log.error("Failed to instantiate header converter class {}; this should have been caught by prior validation logic", headerConverterClass, e);
headerConverterConfigValue.addErrorMessage("Failed to load class " + headerConverterClass + (e.getMessage() != null ? ": " + e.getMessage() : ""));
return null;
}

ConfigDef configDef;
try {
configDef = headerConverter.config();
} catch (RuntimeException e) {
log.error("Failed to load ConfigDef from header converter of type {}", headerConverterClass, e);
headerConverterConfigValue.addErrorMessage("Failed to load ConfigDef from header converter" + (e.getMessage() != null ? ": " + e.getMessage() : ""));
return null;
}
if (configDef == null) {
log.warn("{}.configDef() has returned a null ConfigDef; no further preflight config validation for this converter will be performed", headerConverterClass);
Comment thread
C0urante marked this conversation as resolved.
Outdated
// Older versions of Connect didn't do any header converter validation.
// Even though header converters are technically required to return a non-null ConfigDef object from HeaderConverter::config,
// we permit this case in order to avoid breaking existing header converters that, despite not adhering to this requirement,
// can be used successfully with a connector.
return null;
}

final String headerConverterPrefix = HEADER_CONVERTER_CLASS_CONFIG + ".";
Map<String, String> headerConverterConfig = connectorConfig.entrySet().stream()
.filter(e -> e.getKey().startsWith(headerConverterPrefix))
.collect(Collectors.toMap(
e -> e.getKey().substring(headerConverterPrefix.length()),
Map.Entry::getValue
));
headerConverterConfig.put(ConverterConfig.TYPE_CONFIG, ConverterType.HEADER.getName());

List<ConfigValue> configValues;
try {
configValues = configDef.validate(headerConverterConfig);
} catch (RuntimeException e) {
log.error("Failed to perform custom config validation for header converter of type {}", headerConverterClass, e);
headerConverterConfigValue.addErrorMessage("Failed to perform custom config validation for header converter" + (e.getMessage() != null ? ": " + e.getMessage() : ""));
return null;
}

return prefixedConfigInfos(configDef.configKeys(), configValues, headerConverterPrefix);
}

@Override
public void validateConnectorConfig(Map<String, String> connectorProps, Callback<ConfigInfos> callback) {
validateConnectorConfig(connectorProps, callback, true);
Expand Down Expand Up @@ -426,22 +497,18 @@ ConfigInfos validateConnectorConfig(Map<String, String> connectorProps, boolean
Connector connector = getConnector(connType);
org.apache.kafka.connect.health.ConnectorType connectorType;
ClassLoader savedLoader = plugins().compareAndSwapLoaders(connector);
ConfigDef enrichedConfigDef;
Map<String, ConfigValue> validatedConnectorConfig;
try {
ConfigDef baseConfigDef;
if (connector instanceof SourceConnector) {
baseConfigDef = SourceConnectorConfig.configDef();
connectorType = org.apache.kafka.connect.health.ConnectorType.SOURCE;
enrichedConfigDef = ConnectorConfig.enrich(plugins(), SourceConnectorConfig.configDef(), connectorProps, false);
validatedConnectorConfig = validateSourceConnectorConfig(enrichedConfigDef, connectorProps);
} else {
baseConfigDef = SinkConnectorConfig.configDef();
SinkConnectorConfig.validate(connectorProps);
connectorType = org.apache.kafka.connect.health.ConnectorType.SINK;
enrichedConfigDef = ConnectorConfig.enrich(plugins(), SinkConnectorConfig.configDef(), connectorProps, false);
validatedConnectorConfig = validateSinkConnectorConfig(enrichedConfigDef, connectorProps);
}
ConfigDef enrichedConfigDef = ConnectorConfig.enrich(plugins(), baseConfigDef, connectorProps, false);
Map<String, ConfigValue> validatedConnectorConfig = validateBasicConnectorConfig(
connector,
enrichedConfigDef,
connectorProps
);
List<ConfigValue> configValues = new ArrayList<>(validatedConnectorConfig.values());
Map<String, ConfigKey> configKeys = new LinkedHashMap<>(enrichedConfigDef.configKeys());
Set<String> allGroups = new LinkedHashSet<>(enrichedConfigDef.groups());
Expand All @@ -468,7 +535,11 @@ ConfigInfos validateConnectorConfig(Map<String, String> connectorProps, boolean
configKeys.putAll(configDef.configKeys());
allGroups.addAll(configDef.groups());
configValues.addAll(config.configValues());
ConfigInfos configInfos = generateResult(connType, configKeys, configValues, new ArrayList<>(allGroups));

// do custom header converter-specific validation
ConfigInfos headerConverterConfigInfos = validateHeaderConverterConfig(connectorProps, validatedConnectorConfig.get(HEADER_CONVERTER_CLASS_CONFIG));

ConfigInfos configInfos = generateResult(connType, configKeys, configValues, new ArrayList<>(allGroups));

AbstractConfig connectorConfig = new AbstractConfig(new ConfigDef(), connectorProps, doLog);
String connName = connectorProps.get(ConnectorConfig.NAME_CONFIG);
Expand Down Expand Up @@ -508,7 +579,7 @@ ConfigInfos validateConnectorConfig(Map<String, String> connectorProps, boolean
}

}
return mergeConfigInfos(connType, configInfos, producerConfigInfos, consumerConfigInfos, adminConfigInfos);
return mergeConfigInfos(connType, configInfos, producerConfigInfos, consumerConfigInfos, adminConfigInfos, headerConverterConfigInfos);
} finally {
Plugins.compareAndSwapLoaders(savedLoader);
}
Expand Down Expand Up @@ -536,10 +607,6 @@ private static ConfigInfos validateClientOverrides(String connName,
org.apache.kafka.connect.health.ConnectorType connectorType,
ConnectorClientConfigRequest.ClientType clientType,
ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy) {
int errorCount = 0;
List<ConfigInfo> configInfoList = new LinkedList<>();
Map<String, ConfigKey> configKeys = configDef.configKeys();
Set<String> groups = new LinkedHashSet<>();
Map<String, Object> clientConfigs = new HashMap<>();
for (Map.Entry<String, Object> rawClientConfig : connectorConfig.originalsWithPrefix(prefix).entrySet()) {
String configName = rawClientConfig.getKey();
Expand All @@ -550,30 +617,42 @@ private static ConfigInfos validateClientOverrides(String connName,
: rawConfigValue;
clientConfigs.put(configName, parsedConfigValue);
}

ConnectorClientConfigRequest connectorClientConfigRequest = new ConnectorClientConfigRequest(
connName, connectorType, connectorClass, clientConfigs, clientType);
List<ConfigValue> configValues = connectorClientConfigOverridePolicy.validate(connectorClientConfigRequest);
if (configValues != null) {
for (ConfigValue validatedConfigValue : configValues) {
ConfigKey configKey = configKeys.get(validatedConfigValue.name());
ConfigKeyInfo configKeyInfo = null;
if (configKey != null) {
if (configKey.group != null) {
groups.add(configKey.group);
}
configKeyInfo = convertConfigKey(configKey, prefix);
}

ConfigValue configValue = new ConfigValue(prefix + validatedConfigValue.name(), validatedConfigValue.value(),
validatedConfigValue.recommendedValues(), validatedConfigValue.errorMessages());
if (configValue.errorMessages().size() > 0) {
errorCount++;
return prefixedConfigInfos(configDef.configKeys(), configValues, prefix);
}

private static ConfigInfos prefixedConfigInfos(Map<String, ConfigKey> configKeys, List<ConfigValue> configValues, String prefix) {
int errorCount = 0;
Set<String> groups = new LinkedHashSet<>();
List<ConfigInfo> configInfos = new ArrayList<>();

if (configValues == null) {
return new ConfigInfos("", errorCount, new ArrayList<>(groups), configInfos);
}

for (ConfigValue validatedConfigValue : configValues) {
ConfigKey configKey = configKeys.get(validatedConfigValue.name());
ConfigKeyInfo configKeyInfo = null;
if (configKey != null) {
if (configKey.group != null) {
groups.add(configKey.group);
}
ConfigValueInfo configValueInfo = convertConfigValue(configValue, configKey != null ? configKey.type : null);
configInfoList.add(new ConfigInfo(configKeyInfo, configValueInfo));
configKeyInfo = convertConfigKey(configKey, prefix);
}

ConfigValue configValue = new ConfigValue(prefix + validatedConfigValue.name(), validatedConfigValue.value(),
validatedConfigValue.recommendedValues(), validatedConfigValue.errorMessages());
if (configValue.errorMessages().size() > 0) {
errorCount++;
}
ConfigValueInfo configValueInfo = convertConfigValue(configValue, configKey != null ? configKey.type : null);
configInfos.add(new ConfigInfo(configKeyInfo, configValueInfo));
}
return new ConfigInfos(connectorClass.toString(), errorCount, new ArrayList<>(groups), configInfoList);
return new ConfigInfos("", errorCount, new ArrayList<>(groups), configInfos);
}

// public for testing
Expand Down
Loading