@SuppressWarnings("unchecked")
    protected <T> SortedSet<PluginDesc<T>> getServiceLoaderPluginDesc(PluginType type, PluginSource source) {
        SortedSet<PluginDesc<T>> result = new TreeSet<>();
        ServiceLoader<T> serviceLoader = ServiceLoader.load((Class<T>) type.superClass(), source.loader());
        Iterator<T> iterator = serviceLoader.iterator();
        while (handleLinkageError(type, source, iterator::hasNext)) {
            try (LoaderSwap loaderSwap = withClassLoader(source.loader())) {
                T pluginImpl;
                try {
                    pluginImpl = handleLinkageError(type, source, iterator::next);
                } catch (ServiceConfigurationError t) {
                    log.error("Failed to discover {} in {}{}",
                            type.simpleName(), source, reflectiveErrorDescription(t.getCause()), t);
                    continue;
                }
                Class<? extends T> pluginKlass = (Class<? extends T>) pluginImpl.getClass();
                if (pluginKlass.getClassLoader() != source.loader()) {
                    log.debug("{} from other classloader {} is visible from {}, excluding to prevent isolated loading",
                            type.simpleName(), pluginKlass.getClassLoader(), source);
                    continue;
                }
                result.add(pluginDesc(pluginKlass, versionFor(pluginImpl), type, source));
            }
        }
        return result;
    }
Implementation-specific strategy for scanning a single {@link PluginSource}.
@param source A single source to scan for plugins.
@return A {@link PluginScanResult} containing all plugins which this scanning implementation could discover.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginScanner.java