It works ish

Author: holdenk

core/src/main/java/org/apache/spark/util/ParentClassLoader.java

@@ -18,14 +18,15 @@
 package org.apache.spark.util;
 
 /**
- * A class loader which makes findClass accesiable to the child
+ * A class loader which makes findClass accesible to the child
  */
 public class ParentClassLoader extends ClassLoader {
-    public ParentClassLoader(ClassLoader parent) {
-	super(parent);
+    public ParentClassLoader(ClassLoader myParent) {
+	super(myParent);
     }
+
     @Override
     public Class<?> findClass(String name) throws ClassNotFoundException {
-	return super.loadClass(name);
+	return super.findClass(name);
     }
 }

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -291,16 +291,19 @@ private[spark] class Executor(
    * Create a ClassLoader for use in tasks, adding any JARs specified by the user or any classes
    * created by the interpreter to the search path
    */
-  private def createClassLoader(): ExecutorURLClassLoader = {
-    val loader = Thread.currentThread().getContextClassLoader
+  private def createClassLoader(): AddableURLClassLoader = {
+    val loader = this.getClass.getClassLoader
 
     // For each of the jars in the jarSet, add them to the class loader.
     // We assume each of the files has already been fetched.
     val urls = currentJars.keySet.map { uri =>
       new File(uri.split("/").last).toURI.toURL
     }.toArray
     val userClassPathFirst = conf.getBoolean("spark.classpath.userClassPathFirst", false)
-    new ExecutorURLClassLoader(urls, loader, userClassPathFirst)
+    userClassPathFirst match {
+      case true => new ChildExecutorURLClassLoader(urls, loader)
+      case false => new ExecutorURLClassLoader(urls, loader)
+    }
   }
 
   /**

core/src/main/scala/org/apache/spark/executor/ExecutorURLClassLoader.scala

@@ -26,8 +26,13 @@ import org.apache.spark.util.ParentClassLoader
  * We also make changes so user classes can come before the default classes.
  */
 
-private[spark] class ExecutorURLClassLoader(urls: Array[URL], parent: ClassLoader, userFirst: Boolean)
-  extends ClassLoader {
+trait AddableURLClassLoader extends ClassLoader {
+  def addURL(url: URL)
+  def getURLs(): Array[URL]
+}
+
+private[spark] class ChildExecutorURLClassLoader(urls: Array[URL], parent: ClassLoader)
+  extends ClassLoader with AddableURLClassLoader {
 
   object userClassLoader extends URLClassLoader(urls, null){
     override def addURL(url: URL) {
@@ -41,22 +46,11 @@ private[spark] class ExecutorURLClassLoader(urls: Array[URL], parent: ClassLoade
   val parentClassLoader = new ParentClassLoader(parent)
 
   override def findClass(name: String): Class[_] = {
-    if (!userFirst) {
-      try {
-        val c = parentClassLoader.findClass(name)
-        c
-      } catch {
-        case e: ClassNotFoundException => {
-          userClassLoader.findClass(name)
-        }
-      }
-    } else {
-      try {
-        userClassLoader.findClass(name)
-      } catch {
-        case e: ClassNotFoundException => {
-          parentClassLoader.findClass(name)
-        }
+    try {
+      userClassLoader.findClass(name)
+    } catch {
+      case e: ClassNotFoundException => {
+        parentClassLoader.findClass(name)
       }
     }
   }
@@ -69,3 +63,12 @@ private[spark] class ExecutorURLClassLoader(urls: Array[URL], parent: ClassLoade
     userClassLoader.getURLs()
   }
 }
+
+private[spark] class ExecutorURLClassLoader(urls: Array[URL], parent: ClassLoader)
+  extends URLClassLoader(urls, parent) with AddableURLClassLoader {
+
+  override def addURL(url: URL) {
+    super.addURL(url)
+  }
+}
+

core/src/test/scala/org/apache/spark/executor/ExecutorURLClassLoaderSuite.scala

@@ -28,16 +28,16 @@ class ExecutorURLClassLoaderSuite extends FunSuite {
   val urls = List(new File(spark_home + "/core/src/test/resources/fake-spark-class.jar").toURI.toURL).toArray
   val urls2 = List(new File(spark_home + "/core/src/test/resources/fake-spark-class-2.jar").toURI.toURL).toArray
   test("child first") {
-    val parentLoader = new ExecutorURLClassLoader(urls2, null, false)
-    val classLoader = new ExecutorURLClassLoader(urls, parentLoader, true)
+    val parentLoader = new URLClassLoader(urls2, null)
+    val classLoader = new ChildExecutorURLClassLoader(urls, parentLoader)
     val fakeClass = classLoader.loadClass("org.apache.spark.test.FakeClass2").newInstance()
     val fakeClassVersion = fakeClass.toString
     assert(fakeClassVersion === "1")
   }
 
   test("parent first") {
     val parentLoader = new URLClassLoader(urls2, null)
-    val classLoader = new ExecutorURLClassLoader(urls, parentLoader, false)
+    val classLoader = new ExecutorURLClassLoader(urls, parentLoader)
     val fakeClass = classLoader.loadClass("org.apache.spark.test.FakeClass1").newInstance()
     val fakeClassVersion = fakeClass.toString
     assert(fakeClassVersion === "2")