KAFKA-13328, KAFKA-13329 (2): Add custom preflight validation support for connector header, key, and value converters #14309
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -52,6 +52,8 @@ | |||||
| import org.apache.kafka.connect.storage.ClusterConfigState; | ||||||
| import org.apache.kafka.connect.storage.ConfigBackingStore; | ||||||
| import org.apache.kafka.connect.storage.Converter; | ||||||
| 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.transforms.Transformation; | ||||||
|
|
@@ -79,6 +81,7 @@ | |||||
| import java.util.List; | ||||||
| import java.util.Locale; | ||||||
| import java.util.Map; | ||||||
| import java.util.Objects; | ||||||
| import java.util.Optional; | ||||||
| import java.util.Set; | ||||||
| import java.util.concurrent.ConcurrentHashMap; | ||||||
|
|
@@ -90,6 +93,10 @@ | |||||
| import java.util.regex.Pattern; | ||||||
| import java.util.stream.Collectors; | ||||||
|
|
||||||
| import static org.apache.kafka.connect.runtime.ConnectorConfig.HEADER_CONVERTER_CLASS_CONFIG; | ||||||
| import static org.apache.kafka.connect.runtime.ConnectorConfig.KEY_CONVERTER_CLASS_CONFIG; | ||||||
| import static org.apache.kafka.connect.runtime.ConnectorConfig.VALUE_CONVERTER_CLASS_CONFIG; | ||||||
|
|
||||||
| /** | ||||||
| * Abstract Herder implementation which handles connector/task lifecycle tracking. Extensions | ||||||
| * must invoke the lifecycle hooks appropriately. | ||||||
|
|
@@ -392,6 +399,161 @@ protected Map<String, ConfigValue> validateSourceConnectorConfig(SourceConnector | |||||
| return configDef.validateAll(config); | ||||||
| } | ||||||
|
|
||||||
| /** | ||||||
| * General-purpose validation logic for converters that are configured directly | ||||||
| * in a connector config (as opposed to inherited from the worker config). | ||||||
| * @param connectorConfig the configuration for the connector; may not be null | ||||||
| * @param pluginConfigValue the {@link ConfigValue} for the converter property in the connector config; | ||||||
| * 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 pluginInterface the interface for the plugin type | ||||||
| * (e.g., {@code org.apache.kafka.connect.storage.Converter.class}); | ||||||
| * may not be null | ||||||
| * @param configDefAccessor an accessor that can be used to retrieve a {@link ConfigDef} | ||||||
| * from an instance of the plugin type (e.g., {@code Converter::config}); | ||||||
| * may not be null | ||||||
| * @param pluginName a lowercase, human-readable name for the type of plugin (e.g., {@code "key converter"}); | ||||||
| * may not be null | ||||||
| * @param pluginProperty the property used to define a custom class for the plugin type | ||||||
| * in a connector config (e.g., {@link ConnectorConfig#KEY_CONVERTER_CLASS_CONFIG}); | ||||||
| * may not be null | ||||||
| * @param defaultProperties any default properties to include in the configuration that will be used for | ||||||
| * the plugin; may be null | ||||||
|
|
||||||
| * @return a {@link ConfigInfos} object containing validation results for the plugin in the connector config, | ||||||
| * or null if either no custom validation was performed (possibly because no custom plugin was defined in the | ||||||
| * connector config), or if custom validation failed | ||||||
|
|
||||||
| * @param <T> the plugin class to perform validation for | ||||||
| */ | ||||||
| private <T> ConfigInfos validateConverterConfig( | ||||||
| Map<String, String> connectorConfig, | ||||||
| ConfigValue pluginConfigValue, | ||||||
| Class<T> pluginInterface, | ||||||
| Function<T, ConfigDef> configDefAccessor, | ||||||
| String pluginName, | ||||||
| String pluginProperty, | ||||||
| Map<String, String> defaultProperties, | ||||||
| Function<String, TemporaryStage> reportStage | ||||||
| ) { | ||||||
| Objects.requireNonNull(connectorConfig); | ||||||
| Objects.requireNonNull(pluginInterface); | ||||||
| Objects.requireNonNull(configDefAccessor); | ||||||
| Objects.requireNonNull(pluginName); | ||||||
| Objects.requireNonNull(pluginProperty); | ||||||
|
|
||||||
| String pluginClass = connectorConfig.get(pluginProperty); | ||||||
|
|
||||||
| if (pluginClass == null | ||||||
| || pluginConfigValue == null | ||||||
| || !pluginConfigValue.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. | ||||||
| return null; | ||||||
| } | ||||||
|
|
||||||
| T pluginInstance; | ||||||
| String stageDescription = "instantiating the connector's " + pluginName + " for validation"; | ||||||
| try (TemporaryStage stage = reportStage.apply(stageDescription)) { | ||||||
| pluginInstance = Utils.newInstance(pluginClass, pluginInterface); | ||||||
|
There was a problem hiding this comment. This initializes the plugin with the wrong classloader (and I guess applies the configDefAccessor with the wrong classloader too). There's already Plugins#newPlugin(String) which you could use, but it returns Object. Perhaps we could add a Class argument to make the type signature better. There was a problem hiding this comment. I think this is actually correct. All calls to |
||||||
| } catch (ClassNotFoundException | RuntimeException e) { | ||||||
|
There was a problem hiding this comment.
Suggested change
|
||||||
| 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() : "")); | ||||||
|
There was a problem hiding this comment. I think mutating the passed-in value is correct here, but it's not clear from the signature that this happens. There was a problem hiding this comment. Ack, added a note to the Javadoc. |
||||||
| return null; | ||||||
| } | ||||||
|
|
||||||
| try { | ||||||
| ConfigDef configDef; | ||||||
| stageDescription = "retrieving the configuration definition from the connector's " + pluginName; | ||||||
| try (TemporaryStage stage = reportStage.apply(stageDescription)) { | ||||||
| configDef = configDefAccessor.apply(pluginInstance); | ||||||
| } catch (RuntimeException e) { | ||||||
|
There was a problem hiding this comment.
Suggested change
|
||||||
| log.error("Failed to load ConfigDef from {} of type {}", pluginName, pluginClass, e); | ||||||
| pluginConfigValue.addErrorMessage("Failed to load ConfigDef from " + pluginName + (e.getMessage() != null ? ": " + e.getMessage() : "")); | ||||||
| return null; | ||||||
| } | ||||||
| if (configDef == null) { | ||||||
| log.warn("{}.config() has returned a null ConfigDef; no further preflight config validation for this converter will be performed", pluginClass); | ||||||
| // Older versions of Connect didn't do any converter validation. | ||||||
| // Even though converters are technically required to return a non-null ConfigDef object from their config() method, | ||||||
| // we permit this case in order to avoid breaking existing converters that, despite not adhering to this requirement, | ||||||
| // can be used successfully with a connector. | ||||||
| return null; | ||||||
| } | ||||||
| final String pluginPrefix = pluginProperty + "."; | ||||||
| Map<String, String> pluginConfig = Utils.entriesWithPrefix(connectorConfig, pluginPrefix); | ||||||
| if (defaultProperties != null) | ||||||
| defaultProperties.forEach(pluginConfig::putIfAbsent); | ||||||
|
|
||||||
| List<ConfigValue> configValues; | ||||||
| stageDescription = "performing config validation for the connector's " + pluginName; | ||||||
| try (TemporaryStage stage = reportStage.apply(stageDescription)) { | ||||||
| configValues = configDef.validate(pluginConfig); | ||||||
| } catch (RuntimeException e) { | ||||||
|
There was a problem hiding this comment.
Suggested change
There was a problem hiding this comment. I don't understand this suggestion. When could a checked exception be thrown? There was a problem hiding this comment. For this and the Utils.newInstance comment i was thinking about preparing for #13185 but I suppose these can be addressed in that PR. There was a problem hiding this comment. Got it--yeah, I'd prefer to leave as-is to reduce FUD in case #13185 doesn't get merged soon. |
||||||
| log.error("Failed to perform config validation for {} of type {}", pluginName, pluginClass, e); | ||||||
| pluginConfigValue.addErrorMessage("Failed to perform config validation for " + pluginName + (e.getMessage() != null ? ": " + e.getMessage() : "")); | ||||||
| return null; | ||||||
| } | ||||||
|
|
||||||
| return prefixedConfigInfos(configDef.configKeys(), configValues, pluginPrefix); | ||||||
| } finally { | ||||||
| Utils.maybeCloseQuietly(pluginInstance, pluginName + " " + pluginClass); | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| private ConfigInfos validateHeaderConverterConfig( | ||||||
| Map<String, String> connectorConfig, | ||||||
| ConfigValue headerConverterConfigValue, | ||||||
| Function<String, TemporaryStage> reportStage | ||||||
| ) { | ||||||
| return validateConverterConfig( | ||||||
| connectorConfig, | ||||||
| headerConverterConfigValue, | ||||||
| HeaderConverter.class, | ||||||
| HeaderConverter::config, | ||||||
| "header converter", | ||||||
| HEADER_CONVERTER_CLASS_CONFIG, | ||||||
| Collections.singletonMap(ConverterConfig.TYPE_CONFIG, ConverterType.HEADER.getName()), | ||||||
| reportStage | ||||||
| ); | ||||||
| } | ||||||
|
|
||||||
| 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, | ||||||
| Collections.singletonMap(ConverterConfig.TYPE_CONFIG, ConverterType.KEY.getName()), | ||||||
| reportStage | ||||||
| ); | ||||||
| } | ||||||
|
|
||||||
| 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, | ||||||
| Collections.singletonMap(ConverterConfig.TYPE_CONFIG, ConverterType.VALUE.getName()), | ||||||
| reportStage | ||||||
| ); | ||||||
| } | ||||||
|
|
||||||
| @Override | ||||||
| public void validateConnectorConfig(Map<String, String> connectorProps, Callback<ConfigInfos> callback) { | ||||||
| validateConnectorConfig(connectorProps, callback, true); | ||||||
|
|
@@ -562,8 +724,25 @@ ConfigInfos validateConnectorConfig( | |||||
| 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; | ||||||
|
|
@@ -612,7 +791,15 @@ ConfigInfos validateConnectorConfig( | |||||
| connectorClientConfigOverridePolicy); | ||||||
| } | ||||||
| } | ||||||
| return mergeConfigInfos(connType, | ||||||
| configInfos, | ||||||
| producerConfigInfos, | ||||||
| consumerConfigInfos, | ||||||
| adminConfigInfos, | ||||||
| headerConverterConfigInfos, | ||||||
| keyConverterConfigInfos, | ||||||
| valueConverterConfigInfos | ||||||
| ); | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
|
|
@@ -638,10 +825,6 @@ private static ConfigInfos validateClientOverrides(String connName, | |||||
| org.apache.kafka.connect.health.ConnectorType connectorType, | ||||||
| ConnectorClientConfigRequest.ClientType clientType, | ||||||
| ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy) { | ||||||
| Map<String, Object> clientConfigs = new HashMap<>(); | ||||||
| for (Map.Entry<String, Object> rawClientConfig : connectorConfig.originalsWithPrefix(prefix).entrySet()) { | ||||||
| String configName = rawClientConfig.getKey(); | ||||||
|
|
@@ -655,27 +838,38 @@ private static ConfigInfos validateClientOverrides(String connName, | |||||
| ConnectorClientConfigRequest connectorClientConfigRequest = new ConnectorClientConfigRequest( | ||||||
| connName, connectorType, connectorClass, clientConfigs, clientType); | ||||||
| List<ConfigValue> configValues = connectorClientConfigOverridePolicy.validate(connectorClientConfigRequest); | ||||||
|
|
||||||
| 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) { | ||||||
|
There was a problem hiding this comment. I think this null check is only relevant when the value is coming from the overridePolicy.validate, in validateConverterConfig, I think the ConfigDef#validate call will always be non-null. There was a problem hiding this comment.
I acknowledge that this is extremely unlikely, but it seems like this null guard is the best way to handle that scenario as opposed to, e.g., throwing an error and causing a 500 response to be returned. Thoughts? |
||||||
| return new ConfigInfos("", 0, 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); | ||||||
| } | ||||||
| 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("", errorCount, new ArrayList<>(groups), configInfos); | ||||||
|
There was a problem hiding this comment. nit: i don't like this empty string, but i see that it has no overall effect. I wonder if the validator can just return List and compute errorCount/groups at the end. This doesn't have to be done in this PR. There was a problem hiding this comment. Yeah, it's a little weird with the empty string here. Hopefully it's fine for now but if we continue augmenting and refactoring this class I agree that it might be worth changing. |
||||||
| } | ||||||
|
|
||||||
| // public for testing | ||||||
|
|
||||||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is where the connector validation used the tempConnectors cache to re-use the connector objects.
Personally i'm fine with per-call instantiation, but thought it would be worth mentioning.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good point, and definitely worth calling out. I suspect this won't be a resource bottleneck and GC will be sufficient to clean up these instances before they eat up too much memory, but if not then we can certainly look into caching these plugin instances.