You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/04/19 19:57:51 UTC
[GitHub] [kafka] rhauch commented on a change in pull request #10549: KAFKA-8605 log an error message when we detect multiple copies of sam…
rhauch commented on a change in pull request #10549:
URL: https://github.com/apache/kafka/pull/10549#discussion_r616119723
##########
File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java
##########
@@ -196,8 +214,10 @@ private static PluginClassLoader newPluginClassLoader(
pluginLoaders.put(pluginClassName, inner);
// TODO: once versioning is enabled this line should be moved outside this if branch
log.info("Added plugin '{}'", pluginClassName);
+ allAddedPlugins.put(pluginClassName, new ArrayList<>());
}
inner.put(plugin, loader);
+ allAddedPlugins.get(pluginClassName).add(plugin);
Review comment:
We can use `computeIfAbsent(...)` to eliminate the prior newly-added line:
```suggestion
allAddedPlugins.computeIfAbsent(pluginClassName, n -> new ArrayList<>()).add(plugin);
```
##########
File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java
##########
@@ -153,13 +157,26 @@ public PluginClassLoader pluginClassLoader(String name) {
if (inner == null) {
return null;
}
- ClassLoader pluginLoader = inner.get(inner.lastKey());
+ ClassLoader pluginLoader = inner.get(pluginDescInUse(inner));
return pluginLoader instanceof PluginClassLoader
? (PluginClassLoader) pluginLoader
: null;
}
- public ClassLoader connectorLoader(Connector connector) {
+ //visible for testing
+ PluginDesc<?> pluginDescInUse(String name) {
Review comment:
I realize that `pluginDesc` in this name is just the equivalent of `PluginDesc`, but generally we try to avoid abbreviations, which is probably more true in this case because the name seems even more mangled:
```suggestion
PluginDesc<?> usedPluginDesc(String name) {
```
##########
File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java
##########
@@ -153,13 +157,26 @@ public PluginClassLoader pluginClassLoader(String name) {
if (inner == null) {
return null;
}
- ClassLoader pluginLoader = inner.get(inner.lastKey());
+ ClassLoader pluginLoader = inner.get(pluginDescInUse(inner));
return pluginLoader instanceof PluginClassLoader
? (PluginClassLoader) pluginLoader
: null;
}
- public ClassLoader connectorLoader(Connector connector) {
+ //visible for testing
+ PluginDesc<?> pluginDescInUse(String name) {
+ SortedMap<PluginDesc<?>, ClassLoader> inner = pluginLoaders.get(name);
+ if (inner == null) {
+ return null;
+ }
+ return pluginDescInUse(inner);
+ }
+
+ private PluginDesc<?> pluginDescInUse(SortedMap<PluginDesc<?>, ClassLoader> inner) {
+ return inner.lastKey();
+ }
+
+ public ClassLoader connectorLoader(Connector connector) {
Review comment:
```suggestion
public ClassLoader connectorLoader(Connector connector) {
```
##########
File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java
##########
@@ -208,6 +228,22 @@ protected void initLoaders() {
// Finally add parent/system loader.
initPluginLoader(CLASSPATH_NAME);
addAllAliases();
+ reportPluginConflicts();
+ }
+
+ //visible for testing
+ Set<String> reportPluginConflicts() {
+ Set<String> conflictPluginClasses = new HashSet<>();
+ for (Map.Entry<String, List<PluginDesc<?>>> entry : allAddedPlugins.entrySet()) {
+ String pluginClassName = entry.getKey();
+ List<PluginDesc<?>> pluginDescriptors = entry.getValue();
+ if (pluginDescriptors.size() > 1) {
+ PluginDesc<?> pluginDescInUse = pluginDescInUse(pluginClassName);
+ log.error("For plugin '{}', detected multiple copies '{}', this copy '{}' will be used.", pluginClassName, pluginDescriptors, pluginDescInUse);
+ conflictPluginClasses.add(pluginClassName);
+ }
+ }
+ return conflictPluginClasses;
Review comment:
This could be rewritten a bit more compactly and a bit more functionally:
```suggestion
return allAddedPlugins.entrySet().stream().filter(e -> e.getValue().size() > 1).map(e -> {
String pluginClassName = e.getKey();
PluginDesc<?> pluginDescInUse = pluginDescInUse(pluginClassName);
List<PluginDesc<?>> ignoredPlugins = new ArrayList<>(e.getValue());
ignoredPlugins.remove(pluginDescInUse);
log.error("Detected multiple plugins contain '{}'; using {} and ignoring {}", pluginClassName, pluginDescInUse, ignoredPlugins);
return pluginClassName;
}).collect(Collectors.toSet());
```
My suggestion also includes a reworded error message to put the more meaningful information near the front, and computes the unused plugins and includes them at the end of the message, just in case there are multiple.
##########
File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java
##########
@@ -196,8 +214,10 @@ private static PluginClassLoader newPluginClassLoader(
pluginLoaders.put(pluginClassName, inner);
// TODO: once versioning is enabled this line should be moved outside this if branch
log.info("Added plugin '{}'", pluginClassName);
+ allAddedPlugins.put(pluginClassName, new ArrayList<>());
Review comment:
If we use `computeIfAbsent(...)` below, we don't need this line:
```suggestion
```
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org