fix(resolver):Fix classloader issue for Spark/isolated environments

resolver/src/main/java/com/facebook/airlift/resolver/ArtifactResolver.java

@@ -90,6 +90,16 @@ public class ArtifactResolver
             .add("http://repo.maven.apache.org/maven2/")
             .build();
 
+    private static final String MAVEN_REPOSITORY_SYSTEM_CLASS = "org.apache.maven.repository.RepositorySystem";
+    private static final String PLEXUS_CONTAINER_CLASS = "org.codehaus.plexus.DefaultPlexusContainer";
+    private static final String AETHER_REPOSITORY_SYSTEM_CLASS = "org.eclipse.aether.RepositorySystem";
+
+    /**
+     * Cached effective ClassLoader to avoid repeated Class.forName checks on every container() call.
+     * Initialized lazily on first access and reused thereafter.
+     */
+    private static volatile ClassLoader cachedEffectiveClassLoader;
+
     private final RepositorySystem repositorySystem;
     private final DefaultRepositorySystemSession repositorySystemSession;
     private final List<RemoteRepository> repositories;
@@ -327,7 +337,11 @@ private static PlexusContainer container()
     {
         // TODO: move off Plexus DI, use Sisu instead
         try {
-            ClassWorld classWorld = new ClassWorld("plexus.core", Thread.currentThread().getContextClassLoader());
+            // Fix for Spark and other isolated classloader environments:
+            // Select an effective ClassLoader by checking multiple candidates to ensure Maven Resolver/Plexus components are discoverable at runtime.
+            ClassLoader classLoader = getEffectiveClassLoader();
+
+            ClassWorld classWorld = new ClassWorld("plexus.core", classLoader);
 
             ContainerConfiguration cc = new DefaultContainerConfiguration()
                     .setClassWorld(classWorld)
@@ -350,4 +364,95 @@ private static PlexusContainer container()
             throw new RuntimeException("Error loading Maven system", e);
         }
     }
+
+    /**
+     * Determine the most appropriate ClassLoader for initializing Maven Resolver / Plexus components.
+     *
+     * In distributed or containerized environments (e.g., Spark, Docker), the Thread Context ClassLoader
+     * may not have visibility into all required Maven components due to classloader isolation. This can
+     * lead to ServiceLoader or Plexus lookup failures (e.g., RepositorySystem initialization errors).
+     *
+     * This method attempts multiple classloaders in a deterministic order and selects the first one
+     * that can successfully load required Maven components:
+     *
+     * Priority order:
+     *   1) Thread Context ClassLoader
+     *      - Standard mechanism in typical JVM / Maven execution environments
+     *      - Preferred for test environments where transitive dependencies are loaded
+     *
+     *   2) ClassLoader that loaded this class
+     *      - Most reliable in shaded, embedded, or isolated runtime environments (e.g., Spark executors)
+     *
+     *   3) System ClassLoader
+     *      - Fallback for environments where dependencies are available globally
+     *
+     * The result is cached after the first successful determination to avoid repeated Class.forName
+     * checks on every container() call, which can be expensive in hot paths.
+     *
+     * If none of the candidates pass validation, the method returns the System ClassLoader as the
+     * most reliable fallback, ensuring a non-null ClassLoader is always returned.
+     *
+     * This defensive selection logic helps ensure consistent component discovery across different
+     * runtime environments without relying on a single classloader assumption.
+     *
+     * @return the classloader to use for Maven component discovery (never null)
+     */
+    private static ClassLoader getEffectiveClassLoader()
+    {
+        // Return cached value if already determined
+        if (cachedEffectiveClassLoader != null) {
+            return cachedEffectiveClassLoader;
+        }
+
+        ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
+        if (contextClassLoader != null && canLoadMavenComponents(contextClassLoader)) {
+            cachedEffectiveClassLoader = contextClassLoader;
+            return cachedEffectiveClassLoader;
+        }
+
+        ClassLoader thisClassLoader = ArtifactResolver.class.getClassLoader();
+        if (thisClassLoader != null && canLoadMavenComponents(thisClassLoader)) {
+            cachedEffectiveClassLoader = thisClassLoader;
+            return cachedEffectiveClassLoader;
+        }
+
+        ClassLoader systemClassLoader = ClassLoader.getSystemClassLoader();
+        if (systemClassLoader != null && canLoadMavenComponents(systemClassLoader)) {
+            cachedEffectiveClassLoader = systemClassLoader;
+            return cachedEffectiveClassLoader;
+        }
+
+        ClassLoader fallback = systemClassLoader != null ? systemClassLoader : thisClassLoader;
+        if (fallback == null) {
+            throw new IllegalStateException("Unable to determine a valid ClassLoader for Maven component initialization. " +
+                    "All ClassLoader candidates (context, class, system) are null or cannot load required Maven components.");
+        }
+        cachedEffectiveClassLoader = fallback;
+
+        return cachedEffectiveClassLoader;
+    }
+
+    /**
+     * Verify whether the provided ClassLoader has visibility into essential Maven
+     * and Plexus components required for resolver initialization.
+     *
+     * @param classLoader the ClassLoader to validate
+     * @return true if required Maven and Plexus classes are visible to the ClassLoader;
+     * false otherwise
+     */
+    private static boolean canLoadMavenComponents(ClassLoader classLoader)
+    {
+        try {
+            // Verify visibility of a core Maven repository component
+            Class.forName(MAVEN_REPOSITORY_SYSTEM_CLASS, false, classLoader);
+            // Verify visibility of the Plexus container used for component discovery
+            Class.forName(PLEXUS_CONTAINER_CLASS, false, classLoader);
+            // Verify visibility of Maven Resolver API (critical for dependency resolution)
+            Class.forName(AETHER_REPOSITORY_SYSTEM_CLASS, false, classLoader);
+            return true;
+        }
+        catch (ClassNotFoundException e) {
+            return false;
+        }
+    }
 }