KAFKA-9712: Catch and handle exception thrown by reflections scanner (#…

…8289)

This commit works around a bug in version v0.9.12 of the upstream `reflections` library by catching and handling the exception thrown.

The reflections issue is tracked by:
ronmamo/reflections#273

New unit tests were introduced to test the behavior.

* KAFKA-9712: Catch and handle exception thrown by reflections scanner

* Update connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java

Co-Authored-By: Konstantine Karantasis <konstantine@confluent.io>

* Move result initialization back to right before it is used

* Use `java.io.File` in tests

* Fix checkstyle

Co-authored-by: Konstantine Karantasis <konstantine@confluent.io>

Reviewers: Konstantine Karantasis <konstantine@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>

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

@@ -47,6 +47,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Enumeration;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -339,7 +340,14 @@ private <T> Collection<PluginDesc<T>> getPluginDesc(
             Class<T> klass,
             ClassLoader loader
     ) throws InstantiationException, IllegalAccessException {
-        Set<Class<? extends T>> plugins = reflections.getSubTypesOf(klass);
+        Set<Class<? extends T>> plugins;
+        try {
+            plugins = reflections.getSubTypesOf(klass);
+        } catch (ReflectionsException e) {
+            log.debug("Reflections scanner could not find any classes for URLs: " +
+                    reflections.getConfiguration().getUrls(), e);
+            return Collections.emptyList();
+        }
 
         Collection<PluginDesc<T>> result = new ArrayList<>();
         for (Class<? extends T> plugin : plugins) {

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

@@ -17,15 +17,24 @@
 
 package org.apache.kafka.connect.runtime.isolation;
 
-import java.util.Collections;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
 
-import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
 public class DelegatingClassLoaderTest {
 
+    @Rule
+    public TemporaryFolder pluginDir = new TemporaryFolder();
+
     @Test
     public void testWhiteListedManifestResources() {
         assertTrue(
@@ -61,4 +70,62 @@ public void testLoadingPluginClass() throws ClassNotFoundException {
             assertNotNull(classLoader.pluginClassLoader(pluginClassName));
         }
     }
+
+    @Test
+    public void testLoadingInvalidUberJar() throws Exception {
+        pluginDir.newFile("invalid.jar");
+
+        DelegatingClassLoader classLoader = new DelegatingClassLoader(
+            Collections.singletonList(pluginDir.getRoot().getAbsolutePath()));
+        classLoader.initLoaders();
+    }
+
+    @Test
+    public void testLoadingPluginDirContainsInvalidJarsOnly() throws Exception {
+        pluginDir.newFolder("my-plugin");
+        pluginDir.newFile("my-plugin/invalid.jar");
+
+        DelegatingClassLoader classLoader = new DelegatingClassLoader(
+            Collections.singletonList(pluginDir.getRoot().getAbsolutePath()));
+        classLoader.initLoaders();
+    }
+
+    @Test
+    public void testLoadingNoPlugins() throws Exception {
+        DelegatingClassLoader classLoader = new DelegatingClassLoader(
+            Collections.singletonList(pluginDir.getRoot().getAbsolutePath()));
+        classLoader.initLoaders();
+    }
+
+    @Test
+    public void testLoadingPluginDirEmpty() throws Exception {
+        pluginDir.newFolder("my-plugin");
+
+        DelegatingClassLoader classLoader = new DelegatingClassLoader(
+            Collections.singletonList(pluginDir.getRoot().getAbsolutePath()));
+        classLoader.initLoaders();
+    }
+
+    @Test
+    public void testLoadingMixOfValidAndInvalidPlugins() throws Exception {
+        TestPlugins.assertAvailable();
+
+        pluginDir.newFile("invalid.jar");
+        pluginDir.newFolder("my-plugin");
+        pluginDir.newFile("my-plugin/invalid.jar");
+        Path pluginPath = this.pluginDir.getRoot().toPath();
+
+        for (String sourceJar : TestPlugins.pluginPath()) {
+            Path source = new File(sourceJar).toPath();
+            Files.copy(source, pluginPath.resolve(source.getFileName()));
+        }
+
+        DelegatingClassLoader classLoader = new DelegatingClassLoader(
+            Collections.singletonList(pluginDir.getRoot().getAbsolutePath()));
+        classLoader.initLoaders();
+        for (String pluginClassName : TestPlugins.pluginClasses()) {
+            assertNotNull(classLoader.loadClass(pluginClassName));
+            assertNotNull(classLoader.pluginClassLoader(pluginClassName));
+        }
+    }
 }