gharris1727 commented on code in PR #14055:
URL: https://github.com/apache/kafka/pull/14055#discussion_r1270918323
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java:
##########
@@ -63,16 +67,72 @@ public Plugins(Map<String, String> props) {
// VisibleForTesting
Plugins(Map<String, String> props, ClassLoader parent, ClassLoaderFactory
factory) {
String pluginPath = WorkerConfig.pluginPath(props);
+ PluginDiscoveryMode discoveryMode =
WorkerConfig.pluginDiscovery(props);
List<Path> pluginLocations = PluginUtils.pluginLocations(pluginPath);
delegatingLoader = factory.newDelegatingClassLoader(parent);
Set<PluginSource> pluginSources =
PluginUtils.pluginSources(pluginLocations, delegatingLoader, factory);
- scanResult = initLoaders(pluginSources);
+ scanResult = initLoaders(pluginSources, discoveryMode);
}
- private PluginScanResult initLoaders(Set<PluginSource> pluginSources) {
- PluginScanResult reflectiveScanResult = new
ReflectionScanner().discoverPlugins(pluginSources);
- delegatingLoader.installDiscoveredPlugins(reflectiveScanResult);
- return reflectiveScanResult;
+ public PluginScanResult initLoaders(Set<PluginSource> pluginSources,
PluginDiscoveryMode discoveryMode) {
+ PluginScanResult empty = new PluginScanResult(Collections.emptyList());
+ PluginScanResult serviceLoadingScanResult;
+ try {
+ serviceLoadingScanResult =
PluginDiscoveryMode.serviceLoad(discoveryMode) ?
+ new ServiceLoaderScanner().discoverPlugins(pluginSources)
: empty;
+ } catch (Throwable t) {
+ log.error("Unable to perform ServiceLoader scanning as requested
by {}={}, this error may be avoided by reconfiguring {}={}",
+ WorkerConfig.PLUGIN_DISCOVERY_CONFIG, discoveryMode,
+ WorkerConfig.PLUGIN_DISCOVERY_CONFIG,
PluginDiscoveryMode.ONLY_SCAN, t);
+ throw t;
+ }
+ PluginScanResult reflectiveScanResult =
PluginDiscoveryMode.reflectivelyScan(discoveryMode) ?
+ new ReflectionScanner().discoverPlugins(pluginSources) : empty;
+ PluginScanResult scanResult = new
PluginScanResult(Arrays.asList(reflectiveScanResult, serviceLoadingScanResult));
+ maybeReportHybridDiscoveryIssue(discoveryMode,
serviceLoadingScanResult, scanResult);
+ delegatingLoader.installDiscoveredPlugins(scanResult);
+ return scanResult;
+ }
+
+ private static void maybeReportHybridDiscoveryIssue(PluginDiscoveryMode
discoveryMode, PluginScanResult serviceLoadingScanResult, PluginScanResult
mergedResult) {
+ SortedSet<PluginDesc<?>> missingPlugins = new TreeSet<>();
+ mergedResult.forEach(missingPlugins::add);
+ serviceLoadingScanResult.forEach(missingPlugins::remove);
+ if (missingPlugins.isEmpty()) {
+ switch (discoveryMode) {
+ case ONLY_SCAN:
+ log.debug("Service loading of plugins disabled, consider
reconfiguring {}={}",
+ WorkerConfig.PLUGIN_DISCOVERY_CONFIG,
PluginDiscoveryMode.HYBRID_WARN);
+ break;
+ case HYBRID_WARN:
+ case HYBRID_FAIL:
+ log.warn("All plugins have ServiceLoader manifests,
consider reconfiguring {}={}",
+ WorkerConfig.PLUGIN_DISCOVERY_CONFIG,
PluginDiscoveryMode.SERVICE_LOAD);
+ break;
+ case SERVICE_LOAD:
+ log.debug("Reflective loading of plugins disabled, plugins
without manifests will not be visible");
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown discovery
mode");
+ }
+ } else {
+ String message = String.format(
+ "Plugins are missing ServiceLoader manifests, these
plugins will not be visible with %s=%s: %s",
Review Comment:
> For many of them, this is going to be the first time hearing about new
discovery logic; we should make this message as informative as possible if we
want them to start to make use of it.
> Link to a docs section on the website describing the new plugin discovery
logic
I think to explaining everything in one warning/error message is
unreasonable, and linking out to external documentation is the better strategy.
The error message is already lengthy enough with each plugin listing.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConfig.java:
##########
@@ -122,6 +124,18 @@ public class WorkerConfig extends AbstractConfig {
+ "by the worker's scanner before config providers are initialized
and used to "
+ "replace variables.";
+ public static final String PLUGIN_DISCOVERY_CONFIG = "plugin.discovery";
+ protected static final String PLUGIN_DISCOVERY_DOC = "Method to use to
discover plugins provided in the "
+ + "plugin.path configuration. This can be one of multiple values
with the following meanings:\n"
Review Comment:
No, this mode affects both the classpath and plugin.path similarly. I will
update this comment to reflect that.
--
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
