gharris1727 commented on code in PR #13971:
URL: https://github.com/apache/kafka/pull/13971#discussion_r1264075402
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/PluginScannerTest.java:
##########
@@ -20,79 +20,114 @@
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Set;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+@RunWith(Parameterized.class)
public class PluginScannerTest {
+ private enum ScannerType { Reflection, ServiceLoader };
+
@Rule
public TemporaryFolder pluginDir = new TemporaryFolder();
+ public PluginScanner scanner;
+
+ @Parameterized.Parameters
+ public static Collection<Object[]> parameters() {
+ List<Object[]> values = new ArrayList<>();
+ for (ScannerType type : ScannerType.values()) {
+ values.add(new Object[]{type});
+ }
+ return values;
+ }
+
+ public PluginScannerTest(ScannerType scannerType) {
+ switch (scannerType) {
+ case Reflection:
+ this.scanner = new ReflectionScanner();
+ break;
+ case ServiceLoader:
+ this.scanner = new ServiceLoaderScanner();
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown type " +
scannerType);
+ }
+ }
+
@Test
- public void testLoadingUnloadedPluginClass() {
- DelegatingClassLoader classLoader = initClassLoader(
+ public void testScanningEmptyPluginPath() {
+ PluginScanResult result = scan(
Collections.emptyList()
);
- for (String pluginClassName : TestPlugins.pluginClasses()) {
- assertThrows(ClassNotFoundException.class, () ->
classLoader.loadClass(pluginClassName));
- }
+ assertTrue(result.isEmpty());
}
@Test
- public void testLoadingPluginClass() throws ClassNotFoundException {
- DelegatingClassLoader classLoader = initClassLoader(
+ public void testScanningPluginClasses() {
+ PluginScanResult result = scan(
TestPlugins.pluginPath()
);
+ Set<String> classes = new HashSet<>();
+ result.forEach(pluginDesc -> classes.add(pluginDesc.className()));
for (String pluginClassName : TestPlugins.pluginClasses()) {
- assertNotNull(classLoader.loadClass(pluginClassName));
- assertNotNull(classLoader.pluginClassLoader(pluginClassName));
+ assertTrue("Expected " + pluginClassName + "to be discovered but
it was not",
+ classes.contains(pluginClassName));
Review Comment:
I looked more into how the Reflections library handles this, and it actually
just WARN logs these classes and never shows them to us, so we don't even get
the opportunity to log the error ourselves:
```
[2023-07-14 11:54:26,418] WARN could not get type for name
test.plugins.MissingSuperclassConverter from any class loader
(org.reflections.Reflections:318)
org.reflections.ReflectionsException: could not get type for name
test.plugins.MissingSuperclassConverter
at org.reflections.ReflectionUtils.forName(ReflectionUtils.java:312)
at
org.reflections.ReflectionUtils.lambda$forNames$22(ReflectionUtils.java:330)
at
java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
at
java.base/java.util.HashMap$KeySpliterator.forEachRemaining(HashMap.java:1621)
at
java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
at
java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
at
java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:913)
at
java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at
java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:578)
at org.reflections.ReflectionUtils.forNames(ReflectionUtils.java:332)
at org.reflections.Reflections.getSubTypesOf(Reflections.java:404)
at
org.apache.kafka.connect.runtime.isolation.ReflectionScanner.getPluginDesc(ReflectionScanner.java:118)
at
org.apache.kafka.connect.runtime.isolation.ReflectionScanner.scanPlugins(ReflectionScanner.java:91)
at
org.apache.kafka.connect.runtime.isolation.PluginScanner.scanUrlsAndAddPlugins(PluginScanner.java:78)
at
org.apache.kafka.connect.runtime.isolation.PluginScanner.discoverPlugins(PluginScanner.java:66)
Caused by: java.lang.NoClassDefFoundError: test/plugins/NonExistentInterface
```
I think this takes option 4 off the table, as the reflection-loaded plugins
have never failed due to LinkageErrors, before or after KAFKA-14649 change. The
service-loaded plugins have always failed due to LinkageErrors.
I really like your implementation; it allows ServiceLoaders which make
progress during exceptional hasNext calls to proceed to find other
implementations and prevent shadowing. I'll tweak it a little to make it
accommodate multiple plugins failing with LinkageErrors before triggering
fallback behavior.
And I think I can accept re-throwing the error as a fallback behavior for
ServiceLoaderScanner, if it only appears in situations where:
1. The JVM's implementation of ServiceLoader's Iterator's hasNext does not
make progress on LinkageError
2. A plugin is installed which throws a LinkageError
3. plugin.discovery is in HYBRID_WARN, HYBRID_FAIL, or SERVICE_LOAD.
The first two conditions should be exceedingly rare, the overall situation
has three different possibilities for remediation if it appears in production,
and it allows us to still use the upstream ServiceLoader implementation. I'm
sold.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
