KAFKA-15030: Add connect-plugin-path command-line tool.

bin/connect-plugin-path.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+# 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.
+
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.ConnectPluginPath "$@"

bin/windows/connect-plugin-path.bat

@@ -0,0 +1,17 @@
+@echo off
+rem Licensed to the Apache Software Foundation (ASF) under one or more
+rem contributor license agreements.  See the NOTICE file distributed with
+rem this work for additional information regarding copyright ownership.
+rem The ASF licenses this file to You under the Apache License, Version 2.0
+rem (the "License"); you may not use this file except in compliance with
+rem the License.  You may obtain a copy of the License at
+rem
+rem     http://www.apache.org/licenses/LICENSE-2.0
+rem
+rem Unless required by applicable law or agreed to in writing, software
+rem distributed under the License is distributed on an "AS IS" BASIS,
+rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+rem See the License for the specific language governing permissions and
+rem limitations under the License.
+
+"%~dp0kafka-run-class.bat" org.apache.kafka.tools.ConnectPluginPath %*

build.gradle

@@ -1858,6 +1858,8 @@ project(':tools') {
   dependencies {
     implementation project(':clients')
     implementation project(':server-common')
+    implementation project(':connect:api')
+    implementation project(':connect:runtime')
     implementation project(':log4j-appender')
     implementation project(':tools:tools-api')
     implementation libs.argparse4j
@@ -1876,6 +1878,9 @@ project(':tools') {
     testImplementation project(':core').sourceSets.test.output
     testImplementation project(':server-common')
     testImplementation project(':server-common').sourceSets.test.output
+    testImplementation project(':connect:api')
+    testImplementation project(':connect:runtime')
+    testImplementation project(':connect:runtime').sourceSets.test.output
     testImplementation libs.junitJupiter
     testImplementation libs.mockitoInline // supports mocking static methods, final classes, etc.
     testImplementation libs.mockitoJunitJupiter // supports MockitoExtension

checkstyle/import-control.xml

@@ -304,6 +304,8 @@
     <allow pkg="org.apache.kafka.clients.producer" />
     <allow pkg="org.apache.kafka.clients.consumer" />
     <allow pkg="org.apache.kafka.test" />
+    <allow pkg="org.apache.kafka.connect.runtime" />
+    <allow pkg="org.apache.kafka.connect.runtime.isolation" />
     <allow pkg="com.fasterxml.jackson" />
     <allow pkg="org.jose4j" />
     <allow pkg="net.sourceforge.argparse4j" />

config/tools-log4j.properties

@@ -19,3 +19,6 @@ log4j.appender.stderr=org.apache.log4j.ConsoleAppender
 log4j.appender.stderr.layout=org.apache.log4j.PatternLayout
 log4j.appender.stderr.layout.ConversionPattern=[%d] %p %m (%c)%n
 log4j.appender.stderr.Target=System.err
+
+# for connect-plugin-path
+log4j.logger.org.reflections=ERROR

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

@@ -20,6 +20,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.lang.reflect.Modifier;
 import java.net.MalformedURLException;
@@ -197,15 +198,23 @@ public static boolean isClassFile(Path path) {
         return path.toString().toLowerCase(Locale.ROOT).endsWith(".class");
     }
 
-    public static Set<Path> pluginLocations(String pluginPath) {
+    public static Set<Path> pluginLocations(String pluginPath, boolean failFast) {
         if (pluginPath == null) {
             return Collections.emptySet();
         }
         String[] pluginPathElements = COMMA_WITH_WHITESPACE.split(pluginPath.trim(), -1);
         Set<Path> pluginLocations = new LinkedHashSet<>();
         for (String path : pluginPathElements) {
             try {
-                Path pluginPathElement = Paths.get(path).toAbsolutePath();
+                Path specifiedPath = Paths.get(path);
+                Path pluginPathElement = specifiedPath.toAbsolutePath();
+                if (!specifiedPath.isAbsolute()) {
+                    log.warn("Plugin path element '{}' is relative, evaluating to {}.",
+                            specifiedPath, pluginPathElement);
+                }
+                if (!Files.exists(pluginPathElement)) {
+                    throw new FileNotFoundException(pluginPathElement.toString());
+                }
                 // Currently 'plugin.paths' property is a list of top-level directories
                 // containing plugins
                 if (Files.isDirectory(pluginPathElement)) {
@@ -214,6 +223,9 @@ public static Set<Path> pluginLocations(String pluginPath) {
                     pluginLocations.add(pluginPathElement);
                 }
             } catch (InvalidPathException | IOException e) {
+                if (failFast) {
+                    throw new RuntimeException(e);
+                }
                 log.error("Could not get listing for plugin path: {}. Ignoring.", path, e);
             }
         }
@@ -332,32 +344,38 @@ public static List<Path> pluginUrls(Path topPath) throws IOException {
     public static Set<PluginSource> pluginSources(Set<Path> pluginLocations, ClassLoader classLoader, PluginClassLoaderFactory factory) {
         Set<PluginSource> pluginSources = new LinkedHashSet<>();
         for (Path pluginLocation : pluginLocations) {
-
             try {
-                List<URL> pluginUrls = new ArrayList<>();
-                for (Path path : pluginUrls(pluginLocation)) {
-                    pluginUrls.add(path.toUri().toURL());
-                }
-                URL[] urls = pluginUrls.toArray(new URL[0]);
-                PluginClassLoader loader = factory.newPluginClassLoader(
-                        pluginLocation.toUri().toURL(),
-                        urls,
-                        classLoader
-                );
-                pluginSources.add(new PluginSource(pluginLocation, loader, urls));
+                pluginSources.add(isolatedPluginSource(pluginLocation, classLoader, factory));
             } catch (InvalidPathException | MalformedURLException e) {
                 log.error("Invalid path in plugin path: {}. Ignoring.", pluginLocation, e);
             } catch (IOException e) {
                 log.error("Could not get listing for plugin path: {}. Ignoring.", pluginLocation, e);
             }
         }
-        List<URL> parentUrls = new ArrayList<>();
-        parentUrls.addAll(ClasspathHelper.forJavaClassPath());
-        parentUrls.addAll(ClasspathHelper.forClassLoader(classLoader.getParent()));
-        pluginSources.add(new PluginSource(PluginSource.CLASSPATH, classLoader.getParent(), parentUrls.toArray(new URL[0])));
+        pluginSources.add(classpathPluginSource(classLoader.getParent()));
         return pluginSources;
     }
 
+    public static PluginSource isolatedPluginSource(Path pluginLocation, ClassLoader parent, PluginClassLoaderFactory factory) throws IOException {
+        List<URL> pluginUrls = new ArrayList<>();
+        for (Path path : pluginUrls(pluginLocation)) {
+            pluginUrls.add(path.toUri().toURL());
+        }
+        URL[] urls = pluginUrls.toArray(new URL[0]);
+        PluginClassLoader loader = factory.newPluginClassLoader(
+                pluginLocation.toUri().toURL(),
+                urls,
+                parent
+        );
+        return new PluginSource(pluginLocation, loader, urls);
+    }
+
+    public static PluginSource classpathPluginSource(ClassLoader classLoader) {
+        List<URL> parentUrls = new ArrayList<>();
+        parentUrls.addAll(ClasspathHelper.forJavaClassPath());
+        parentUrls.addAll(ClasspathHelper.forClassLoader(classLoader));
+        return new PluginSource(PluginSource.CLASSPATH, classLoader, parentUrls.toArray(new URL[0]));
+    }
 
     /**
      * Return the simple class name of a plugin as {@code String}.
@@ -369,6 +387,11 @@ public static String simpleName(PluginDesc<?> plugin) {
         return plugin.pluginClass().getSimpleName();
     }
 
+    public static String simpleName(String fullClassName) {
+        return fullClassName.substring(fullClassName.lastIndexOf('.') + 1);
+    }
+
+
     /**
      * Remove the plugin type name at the end of a plugin class name, if such suffix is present.
      * This method is meant to be used to extract plugin aliases.
@@ -377,18 +400,22 @@ public static String simpleName(PluginDesc<?> plugin) {
      * @return the pruned simple class name of the plugin.
      */
     public static String prunedName(PluginDesc<?> plugin) {
+        return prunedName(plugin.className(), plugin.type());
+    }
+
+    public static String prunedName(String fullClassName, PluginType type) {
         // It's currently simpler to switch on type than do pattern matching.
-        switch (plugin.type()) {
+        switch (type) {
             case SOURCE:
             case SINK:
-                return prunePluginName(plugin, "Connector");
+                return prunePluginName(fullClassName, "Connector");
             default:
-                return prunePluginName(plugin, plugin.type().simpleName());
+                return prunePluginName(fullClassName, type.simpleName());
         }
     }
 
-    private static String prunePluginName(PluginDesc<?> plugin, String suffix) {
-        String simple = plugin.pluginClass().getSimpleName();
+    private static String prunePluginName(String fullClassName, String suffix) {
+        String simple = simpleName(fullClassName);
         int pos = simple.lastIndexOf(suffix);
         if (pos > 0) {
             return simple.substring(0, pos);

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

@@ -63,7 +63,7 @@ public Plugins(Map<String, String> props) {
     // VisibleForTesting
     Plugins(Map<String, String> props, ClassLoader parent, ClassLoaderFactory factory) {
         String pluginPath = WorkerConfig.pluginPath(props);
-        Set<Path> pluginLocations = PluginUtils.pluginLocations(pluginPath);
+        Set<Path> pluginLocations = PluginUtils.pluginLocations(pluginPath, false);
         delegatingLoader = factory.newDelegatingClassLoader(parent);
         Set<PluginSource> pluginSources = PluginUtils.pluginSources(pluginLocations, delegatingLoader, factory);
         scanResult = initLoaders(pluginSources);

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/TestPlugins.java

@@ -173,7 +173,14 @@ public enum TestPlugin {
         /**
          * A ServiceLoader discovered plugin which subclasses another plugin which is present on the classpath
          */
-        SUBCLASS_OF_CLASSPATH_OVERRIDE_POLICY("subclass-of-classpath", "test.plugins.SubclassOfClasspathOverridePolicy");
+        SUBCLASS_OF_CLASSPATH_OVERRIDE_POLICY("subclass-of-classpath", "test.plugins.SubclassOfClasspathOverridePolicy"),
+        NON_MIGRATED_CONVERTER("non-migrated", "test.plugins.NonMigratedConverter", false),
+        NON_MIGRATED_HEADER_CONVERTER("non-migrated", "test.plugins.NonMigratedHeaderConverter", false),
+        NON_MIGRATED_MULTI_PLUGIN("non-migrated", "test.plugins.NonMigratedMultiPlugin", false),
+        NON_MIGRATED_PREDICATE("non-migrated", "test.plugins.NonMigratedPredicate", false),
+        NON_MIGRATED_SINK_CONNECTOR("non-migrated", "test.plugins.NonMigratedSinkConnector", false),
+        NON_MIGRATED_SOURCE_CONNECTOR("non-migrated", "test.plugins.NonMigratedSourceConnector", false),
+        NON_MIGRATED_TRANSFORMATION("non-migrated", "test.plugins.NonMigratedTransformation", false);
 
         private final String resourceDir;
         private final String className;

connect/runtime/src/test/resources/test-plugins/non-migrated/test/plugins/NonMigratedConverter.java

@@ -0,0 +1,46 @@
+/*
+ * 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 test.plugins;
+
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaAndValue;
+import org.apache.kafka.connect.storage.Converter;
+
+import java.util.Map;
+
+/**
+ * Fake plugin class for testing classloading isolation.
+ * See {@link org.apache.kafka.connect.runtime.isolation.TestPlugins}.
+ * <p>Class which is not migrated to include a service loader manifest.
+ */
+public final class NonMigratedConverter implements Converter {
+
+  @Override
+  public void configure(final Map<String, ?> configs, final boolean isKey) {
+  }
+
+  @Override
+  public byte[] fromConnectData(final String topic, final Schema schema, final Object value) {
+    return new byte[0];
+  }
+
+  @Override
+  public SchemaAndValue toConnectData(final String topic, final byte[] value) {
+    return null;
+  }
+}

connect/runtime/src/test/resources/test-plugins/non-migrated/test/plugins/NonMigratedHeaderConverter.java

@@ -0,0 +1,57 @@
+/*
+ * 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 test.plugins;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaAndValue;
+import org.apache.kafka.connect.storage.HeaderConverter;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * Fake plugin class for testing classloading isolation.
+ * See {@link org.apache.kafka.connect.runtime.isolation.TestPlugins}.
+ * <p>Class which is not migrated to include a service loader manifest.
+ */
+public class NonMigratedHeaderConverter implements HeaderConverter {
+
+  @Override
+  public SchemaAndValue toConnectHeader(String topic, String headerKey, byte[] value) {
+    return null;
+  }
+
+  @Override
+  public byte[] fromConnectHeader(String topic, String headerKey, Schema schema, Object value) {
+    return new byte[0];
+  }
+
+  @Override
+  public ConfigDef config() {
+    return null;
+  }
+
+  @Override
+  public void close() throws IOException {
+  }
+
+  @Override
+  public void configure(Map<String, ?> configs) {
+  }
+}