[SPARK-11818][REPL] Fix ExecutorClassLoader to lookup resources from …

…parent class loader

Without patch, two additional tests of ExecutorClassLoaderSuite fails.

- "resource from parent"
- "resources from parent"

Detailed explanation is here, https://issues.apache.org/jira/browse/SPARK-11818?focusedCommentId=15011202&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-15011202

Author: Jungtaek Lim <kabhwan@gmail.com>

Closes #9812 from HeartSaVioR/SPARK-11818.

(cherry picked from commit be9dd15)
Signed-off-by: Marcelo Vanzin <vanzin@cloudera.com>

repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala

@@ -34,7 +34,9 @@ import org.apache.spark.util.ParentClassLoader
 /**
  * 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.
- * Allows the user to specify if user class path should be first
+ * Allows the user to specify if user class path should be first.
+ * This class loader delegates getting/finding resources to parent loader,
+ * which makes sense until REPL never provide resource dynamically.
  */
 class ExecutorClassLoader(conf: SparkConf, classUri: String, parent: ClassLoader,
     userClassPathFirst: Boolean) extends ClassLoader with Logging {
@@ -55,6 +57,14 @@ class ExecutorClassLoader(conf: SparkConf, classUri: String, parent: ClassLoader
     }
   }
 
+  override def getResource(name: String): URL = {
+    parentLoader.getResource(name)
+  }
+
+  override def getResources(name: String): java.util.Enumeration[URL] = {
+    parentLoader.getResources(name)
+  }
+
   override def findClass(name: String): Class[_] = {
     userClassPathFirst match {
       case true => findClassLocally(name).getOrElse(parentLoader.loadClass(name))

repl/src/test/scala/org/apache/spark/repl/ExecutorClassLoaderSuite.scala

@@ -19,8 +19,13 @@ package org.apache.spark.repl
 
 import java.io.File
 import java.net.{URL, URLClassLoader}
+import java.nio.charset.StandardCharsets
+import java.util
+
+import com.google.common.io.Files
 
 import scala.concurrent.duration._
+import scala.io.Source
 import scala.language.implicitConversions
 import scala.language.postfixOps
 
@@ -41,6 +46,7 @@ class ExecutorClassLoaderSuite
 
   val childClassNames = List("ReplFakeClass1", "ReplFakeClass2")
   val parentClassNames = List("ReplFakeClass1", "ReplFakeClass2", "ReplFakeClass3")
+  val parentResourceNames = List("fake-resource.txt")
   var tempDir1: File = _
   var tempDir2: File = _
   var url1: String = _
@@ -54,6 +60,9 @@ class ExecutorClassLoaderSuite
     url1 = "file://" + tempDir1
     urls2 = List(tempDir2.toURI.toURL).toArray
     childClassNames.foreach(TestUtils.createCompiledClass(_, tempDir1, "1"))
+    parentResourceNames.foreach { x =>
+      Files.write("resource".getBytes(StandardCharsets.UTF_8), new File(tempDir2, x))
+    }
     parentClassNames.foreach(TestUtils.createCompiledClass(_, tempDir2, "2"))
   }
 
@@ -99,6 +108,26 @@ class ExecutorClassLoaderSuite
     }
   }
 
+  test("resource from parent") {
+    val parentLoader = new URLClassLoader(urls2, null)
+    val classLoader = new ExecutorClassLoader(new SparkConf(), url1, parentLoader, true)
+    val resourceName: String = parentResourceNames.head
+    val is = classLoader.getResourceAsStream(resourceName)
+    assert(is != null, s"Resource $resourceName not found")
+    val content = Source.fromInputStream(is, "UTF-8").getLines().next()
+    assert(content.contains("resource"), "File doesn't contain 'resource'")
+  }
+
+  test("resources from parent") {
+    val parentLoader = new URLClassLoader(urls2, null)
+    val classLoader = new ExecutorClassLoader(new SparkConf(), url1, parentLoader, true)
+    val resourceName: String = parentResourceNames.head
+    val resources: util.Enumeration[URL] = classLoader.getResources(resourceName)
+    assert(resources.hasMoreElements, s"Resource $resourceName not found")
+    val fileReader = Source.fromInputStream(resources.nextElement().openStream()).bufferedReader()
+    assert(fileReader.readLine().contains("resource"), "File doesn't contain 'resource'")
+  }
+
   test("failing to fetch classes from HTTP server should not leak resources (SPARK-6209)") {
     // This is a regression test for SPARK-6209, a bug where each failed attempt to load a class
     // from the driver's class server would leak a HTTP connection, causing the class server's