diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala
index 1a0cd2e4b7198..2f2f3b34a6fbc 100644
--- a/core/src/main/scala/org/apache/spark/executor/Executor.scala
+++ b/core/src/main/scala/org/apache/spark/executor/Executor.scala
@@ -314,11 +314,12 @@ private[spark] class Executor(
     val classUri = conf.get("spark.repl.class.uri", null)
     if (classUri != null) {
       logInfo("Using REPL class URI: " + classUri)
+      val userClassPathFirst: java.lang.Boolean = conf.getBoolean("spark.classpath.userClassPathFirst", false)
       try {
-        val klass = Class.forName("org.apache.spark.repl.ExecutorClassLoader")
+        val klass = Class.forName("org.apache.spark.repl.FlexibleExecutorClassLoader")
           .asInstanceOf[Class[_ <: ClassLoader]]
-        val constructor = klass.getConstructor(classOf[String], classOf[ClassLoader])
-        constructor.newInstance(classUri, parent)
+        val constructor = klass.getConstructor(classOf[String], classOf[ClassLoader], classOf[Boolean])
+        constructor.newInstance(classUri, parent, userClassPathFirst)
       } catch {
         case _: ClassNotFoundException =>
           logError("Could not find org.apache.spark.repl.ExecutorClassLoader on classpath!")
diff --git a/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala b/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
index 88340ff808948..015acfcb54243 100644
--- a/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
+++ b/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala
@@ -26,7 +26,7 @@ import org.apache.hadoop.fs.{FileSystem, Path}
 
 import org.apache.spark.SparkEnv
 import org.apache.spark.util.Utils
-
+import org.apache.spark.util.ParentClassLoader
 
 import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm._
 import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.Opcodes._
@@ -35,18 +35,17 @@ import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.Opcodes._
  * A ClassLoader that reads classes from a Hadoop FileSystem or HTTP URI,
  * used to load classes defined by the interpreter when the REPL is used
  */
-class ExecutorClassLoader(classUri: String, parent: ClassLoader, userClassPathFirst: Boolean)
+class ExecutorClassLoader(classUri: String, parent: ClassLoader) extends FlexibleExecutorClassLoader(classUri, parent, false) {
+}
+/**
+ * Allows the user to specify if user class path should be first
+ */ 
+class FlexibleExecutorClassLoader(classUri: String, parent: ClassLoader, userClassPathFirst: Boolean)
 extends ClassLoader {
   val uri = new URI(classUri)
   val directory = uri.getPath
 
-  class ParentLoader(parent: ClassLoader) extends ClassLoader(parent) {
-    override def findClass(name: String): Class[_] = {
-      super.findClass(name)
-    }
-  }
-
-  val parentLoader = new ParentLoader(parent)
+  val parentLoader = new ParentClassLoader(parent)
 
   // Hadoop FileSystem object for our URI, if it isn't using HTTP
   var fileSystem: FileSystem = {
diff --git a/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala b/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
index 4874b7931829b..d0f3cd74a2b2b 100644
--- a/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
+++ b/repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala
@@ -30,7 +30,7 @@ class ExecutorClassLoaderSuite extends FunSuite {
 
   test("child first") {
     val parentLoader = new URLClassLoader(urls2, null)
-    val classLoader = new ExecutorClassLoader(url1, parentLoader, true)
+    val classLoader = new FlexibleExecutorClassLoader(url1, parentLoader, true)
     val fakeClass = classLoader.loadClass("org.apache.spark.test.FakeClass2").newInstance()
     val fakeClassVersion = fakeClass.toString
     assert(fakeClassVersion === "1")
@@ -38,7 +38,7 @@ class ExecutorClassLoaderSuite extends FunSuite {
 
   test("parent first") {
     val parentLoader = new URLClassLoader(urls2, null)
-    val classLoader = new ExecutorClassLoader(url1, parentLoader, false)
+    val classLoader = new ExecutorClassLoader(url1, parentLoader)
     val fakeClass = classLoader.loadClass("org.apache.spark.test.FakeClass1").newInstance()
     val fakeClassVersion = fakeClass.toString
     assert(fakeClassVersion === "2")
@@ -46,7 +46,7 @@ class ExecutorClassLoaderSuite extends FunSuite {
 
   test("child first can fall back") {
     val parentLoader = new URLClassLoader(urls2, null)
-    val classLoader = new ExecutorClassLoader(url1, parentLoader, true)
+    val classLoader = new FlexibleExecutorClassLoader(url1, parentLoader, true)
     val fakeClass = classLoader.loadClass("org.apache.spark.test.FakeClass3").newInstance()
     val fakeClassVersion = fakeClass.toString
     assert(fakeClassVersion === "2")
@@ -54,7 +54,7 @@ class ExecutorClassLoaderSuite extends FunSuite {
 
   test("child first can fail") {
     val parentLoader = new URLClassLoader(urls2, null)
-    val classLoader = new ExecutorClassLoader(url1, parentLoader, true)
+    val classLoader = new FlexibleExecutorClassLoader(url1, parentLoader, true)
     intercept[java.lang.ClassNotFoundException] {
       classLoader.loadClass("org.apache.spark.test.FakeClassDoesNotExist").newInstance()
     }