[WIP] KIP-891: Connect Multiversion Support (Transformation and Predicate Changes)

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java

@@ -417,8 +417,8 @@ protected ConfigDef initialConfigDef() {
             }
 
             @Override
-            protected Stream<Map.Entry<String, ConfigDef.ConfigKey>> configDefsForClass(String typeConfig) {
-                return super.configDefsForClass(typeConfig)
+            protected Stream<Map.Entry<String, ConfigDef.ConfigKey>> configDefsForClass(String typeConfig, String versionConfig, Plugins plugins) {
+                return super.configDefsForClass(typeConfig, versionConfig, plugins)
                         .filter(entry -> {
                             // The implicit parameters mask any from the transformer with the same name
                             if (TransformationStage.PREDICATE_CONFIG.equals(entry.getKey())
@@ -447,7 +447,13 @@ protected void validateProps(String prefix) {
                             "but there is no config '" + prefixedPredicate + "' defining a predicate to be negated.");
                 }
             }
-        }.enrich(newDef);
+
+            @Override
+            protected ConfigDef.Recommender versionRecommender(String typeConfig) {
+                return new PluginsRecommenders(plugins).transformationPluginRecommender(typeConfig);
+            }
+
+        }.enrich(newDef, plugins);
 
         new EnrichablePlugin<Predicate<?>>("Predicate", PREDICATES_CONFIG, PREDICATES_GROUP,
                 (Class) Predicate.class, props, requireFullConfig) {
@@ -460,7 +466,14 @@ protected Set<PluginDesc<Predicate<?>>> plugins() {
             protected ConfigDef config(Predicate<?> predicate) {
                 return predicate.config();
             }
-        }.enrich(newDef);
+
+            @Override
+            protected ConfigDef.Recommender versionRecommender(String typeConfig) {
+                return new PluginsRecommenders(plugins).predicatePluginRecommender(typeConfig);
+            }
+
+        }.enrich(newDef, plugins);
+
         return newDef;
     }
 
@@ -530,7 +543,6 @@ private static void updateKeyDefault(ConfigDef configDef, String versionConfigKe
         ));
     }
 
-    @SuppressWarnings("unchecked")
     private static <T> String fetchPluginVersion(Plugins plugins, String connectorClass, String connectorVersion, String pluginName) {
         if (pluginName == null) {
             return null;
@@ -576,7 +588,7 @@ public EnrichablePlugin(
         }
 
         /** Add the configs for this alias to the given {@code ConfigDef}. */
-        void enrich(ConfigDef newDef) {
+        void enrich(ConfigDef newDef, Plugins plugins) {
             Object aliases = ConfigDef.parseType(aliasConfig, props.get(aliasConfig), Type.LIST);
             if (!(aliases instanceof List)) {
                 return;
@@ -594,12 +606,15 @@ void enrich(ConfigDef newDef) {
                 int orderInGroup = 0;
 
                 final String typeConfig = prefix + "type";
+                final String versionConfig = prefix + WorkerConfig.PLUGIN_VERSION_SUFFIX;
+
+                // Add the class configuration
                 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);
+                                getConfigDefFromPlugin(typeConfig, ((Class<?>) value).getName(), null, plugins);
                             }
                         },
                         () -> "valid configs for " + alias + " " + aliasKind.toLowerCase(Locale.ENGLISH));
@@ -608,7 +623,25 @@ void enrich(ConfigDef newDef) {
                         baseClass.getSimpleName() + " type for " + alias,
                         Collections.emptyList(), new ClassRecommender());
 
-                final ConfigDef configDef = populateConfigDef(typeConfig);
+                // Add the version configuration
+                final ConfigDef.Validator versionValidator = (name, value) -> {
+                    try {
+                        PluginUtils.connectorVersionRequirement((String) value);
+                    } catch (InvalidVersionSpecificationException e) {
+                        throw new ConfigException(name, value, e.getMessage());
+                    }
+                    if (value != null) {
+                        getConfigDefFromPlugin(typeConfig, props.get(typeConfig), (String) value, plugins);
+                    }
+                };
+                String defaultVersion = fetchPluginVersion(plugins, props.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG),
+                        props.get(ConnectorConfig.CONNECTOR_VERSION), props.get(typeConfig));
+                newDef.define(versionConfig, Type.STRING, defaultVersion, versionValidator, Importance.HIGH,
+                        "Version of the '" + alias + "' " + aliasKind.toLowerCase(Locale.ENGLISH) + ".", group, orderInGroup++, Width.LONG,
+                        baseClass.getSimpleName() + " version for " + alias,
+                        Collections.emptyList(), versionRecommender(typeConfig));
+
+                final ConfigDef configDef = populateConfigDef(typeConfig, versionConfig, plugins);
                 if (configDef == null) continue;
                 newDef.embed(prefix, group, orderInGroup, configDef);
             }
@@ -622,10 +655,10 @@ protected void validateProps(String prefix) {
          * Populates the ConfigDef according to the configs returned from {@code configs()} method of class
          * named in the {@code ...type} parameter of the {@code props}.
          */
-        protected ConfigDef populateConfigDef(String typeConfig) {
+        protected ConfigDef populateConfigDef(String typeConfig, String versionConfig, Plugins plugins) {
             final ConfigDef configDef = initialConfigDef();
             try {
-                configDefsForClass(typeConfig)
+                configDefsForClass(typeConfig, versionConfig, plugins)
                         .forEach(entry -> configDef.define(entry.getValue()));
             } catch (ConfigException e) {
                 if (requireFullConfig) {
@@ -641,9 +674,8 @@ protected ConfigDef populateConfigDef(String typeConfig) {
          * Return a stream of configs provided by the {@code configs()} method of class
          * named in the {@code ...type} parameter of the {@code props}.
          */
-        protected Stream<Map.Entry<String, ConfigDef.ConfigKey>> configDefsForClass(String typeConfig) {
-            final Class<?> cls = (Class<?>) ConfigDef.parseType(typeConfig, props.get(typeConfig), Type.CLASS);
-            return getConfigDefFromConfigProvidingClass(typeConfig, cls)
+        protected Stream<Map.Entry<String, ConfigDef.ConfigKey>> configDefsForClass(String typeConfig, String versionConfig, Plugins plugins) {
+            return getConfigDefFromPlugin(typeConfig, props.get(typeConfig), props.get(versionConfig), plugins)
                     .configKeys().entrySet().stream();
         }
 
@@ -652,30 +684,33 @@ protected ConfigDef initialConfigDef() {
             return new ConfigDef();
         }
 
-        /**
-         * Return {@link ConfigDef} from {@code cls}, which is expected to be a non-null {@code Class<T>},
-         * by instantiating it and invoking {@link #config(T)}.
-         * @param key
-         * @param cls The subclass of the baseclass.
-         */
-        ConfigDef getConfigDefFromConfigProvidingClass(String key, Class<?> cls) {
-            if (cls == null) {
+        @SuppressWarnings("unchecked")
+        ConfigDef getConfigDefFromPlugin(String key, String pluginClass, String version, Plugins plugins) {
+            if (pluginClass == null) {
                 throw new ConfigException(key, null, "Not a " + baseClass.getSimpleName());
             }
-            Utils.ensureConcreteSubclass(baseClass, cls);
-
-            T pluginInstance;
+            String connectorClass = props.get(CONNECTOR_CLASS_CONFIG);
+            if (connectorClass == null) {
+                throw new ConfigException(key, pluginClass, "Cannot reference the connector class itself.");
+            }
+            String connectorVersion = props.get(CONNECTOR_VERSION);
+            T plugin;
             try {
-                pluginInstance = Utils.newInstance(cls, baseClass);
+                VersionRange connectorVersionRange = PluginUtils.connectorVersionRequirement(connectorVersion);
+                VersionRange pluginVersion = PluginUtils.connectorVersionRequirement(version);
+                plugin = (T) plugins.newPlugin(pluginClass, pluginVersion, plugins.pluginLoader(connectorClass, connectorVersionRange));
+            } catch (InvalidVersionSpecificationException e) {
+                // this should be caught in the validation of the version string
+                return new ConfigDef();
             } catch (Exception e) {
-                throw new ConfigException(key, String.valueOf(cls), "Error getting config definition from " + baseClass.getSimpleName() + ": " + e.getMessage());
+                throw new ConfigException(key, pluginClass, "Error getting config definition from " + baseClass.getSimpleName() + ": " + e.getMessage());
             }
-            ConfigDef configDef = config(pluginInstance);
+            ConfigDef configDef = config(plugin);
             if (null == configDef) {
                 throw new ConnectException(
                         String.format(
                                 "%s.config() must return a ConfigDef that is not null.",
-                                cls.getName()
+                                plugin.getClass().getName()
                         )
                 );
             }
@@ -695,6 +730,8 @@ ConfigDef getConfigDefFromConfigProvidingClass(String key, Class<?> cls) {
          */
         protected abstract Set<PluginDesc<T>> plugins();
 
+        protected abstract ConfigDef.Recommender versionRecommender(String typeConfig);
+
         /**
          * Recommend bundled transformations or predicates.
          */

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java

@@ -270,10 +270,11 @@ public String latestVersion(String classOrAlias) {
     }
 
     public String pluginVersion(String classOrAlias, ClassLoader sourceLoader) {
-        if (!(sourceLoader instanceof PluginClassLoader)) {
-            return latestVersion(classOrAlias);
+        String version = null;
+        if (sourceLoader instanceof PluginClassLoader) {
+            version = delegatingLoader.versionInLocation(classOrAlias, ((PluginClassLoader) sourceLoader).location());
         }
-        return delegatingLoader.versionInLocation(classOrAlias, ((PluginClassLoader) sourceLoader).location());
+        return version != null ? version : latestVersion(classOrAlias);
     }
 
     public DelegatingClassLoader delegatingLoader() {

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginsRecommenders.java

@@ -19,10 +19,13 @@
 
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.connect.runtime.ConnectorConfig;
+import org.apache.kafka.connect.transforms.Transformation;
+import org.apache.kafka.connect.transforms.predicates.Predicate;
 
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -74,6 +77,14 @@
         return headerConverterPluginVersionRecommender;
     }
 
+    public TransformationPluginRecommender transformationPluginRecommender(String classOrAlias) {
+        return new TransformationPluginRecommender(classOrAlias);
+    }
+
+    public PredicatePluginRecommender predicatePluginRecommender(String classOrAlias) {
+        return new PredicatePluginRecommender(classOrAlias);
+    }
+
     public class ConnectorPluginVersionRecommender implements ConfigDef.Recommender {
 
         @SuppressWarnings({"unchecked", "rawtypes"})
@@ -195,4 +206,60 @@
                     .map(PluginDesc::version).distinct().collect(Collectors.toList());
         }
     }
+
+    // Recommender for transformation and predicate plugins
+    public abstract class SMTPluginRecommender<T> implements ConfigDef.Recommender {
+
+        abstract protected Function<String, Set<PluginDesc<T>>> plugins();
+
+        protected final String classOrAliasConfig;
+
+        public SMTPluginRecommender(String classOrAliasConfig) {
+            this.classOrAliasConfig = classOrAliasConfig;
+        }
+
+        @Override
+        @SuppressWarnings({"rawtypes"})
+        public List<Object> validValues(String name, Map<String, Object> parsedConfig) {
+            if (plugins == null) {
+                return Collections.emptyList();
+            }
+            if (parsedConfig.get(classOrAliasConfig) == null) {
+                return Collections.emptyList();
+            }
+
+            Class classOrAlias = (Class) parsedConfig.get(classOrAliasConfig);
+            return plugins().apply(classOrAlias.getName())
+                    .stream().map(PluginDesc::version).distinct().collect(Collectors.toList());
+        }
+
+        @Override
+        public boolean visible(String name, Map<String, Object> parsedConfig) {
+            return true;
+        }
+    }
+
+    public class TransformationPluginRecommender extends SMTPluginRecommender<Transformation<?>> {
+
+        public TransformationPluginRecommender(String classOrAliasConfig) {
+            super(classOrAliasConfig);
+        }
+
+        @Override
+        protected Function<String, Set<PluginDesc<Transformation<?>>>> plugins() {
+            return plugins::transformations;
+        }
+    }
+
+    public class PredicatePluginRecommender extends SMTPluginRecommender<Predicate<?>> {
+
+        public PredicatePluginRecommender(String classOrAliasConfig) {
+            super(classOrAliasConfig);
+        }
+
+        @Override
+        protected Function<String, Set<PluginDesc<Predicate<?>>>> plugins() {
+            return plugins::predicates;
+        }
+    }
 }