[SPARK-25235][SHELL] Merge the REPL code in Scala 2.11 and 2.12 branches

## What changes were proposed in this pull request?

Using some reflection tricks to merge Scala 2.11 and 2.12 codebase.

## How was this patch tested?

Existing tests.

Closes #22246 from dbtsai/repl.

Lead-authored-by: DB Tsai <d_tsai@apple.com>
Co-authored-by: Liang-Chi Hsieh <viirya@gmail.com>
Signed-off-by: DB Tsai <d_tsai@apple.com>

Author: dbtsai

repl/pom.xml

@@ -167,14 +167,4 @@
     </plugins>
   </build>
 
-  <profiles>
-    <profile>
-      <id>scala-2.12</id>
-      <properties>
-        <extra.source.dir>scala-2.12/src/main/scala</extra.source.dir>
-        <extra.testsource.dir>scala-2.12/src/test/scala</extra.testsource.dir>
-      </properties>
-    </profile>
-  </profiles>
-
 </project>

repl/scala-2.12/src/main/scala/org/apache/spark/repl/SparkILoop.scala

@@ -24,7 +24,6 @@ import scala.Predef.{println => _, _}
 // scalastyle:on println
 import scala.concurrent.Future
 import scala.reflect.classTag
-import scala.reflect.internal.util.ScalaClassLoader.savingContextLoader
 import scala.reflect.io.File
 import scala.tools.nsc.{GenericRunnerSettings, Properties}
 import scala.tools.nsc.Settings
@@ -33,7 +32,7 @@ import scala.tools.nsc.interpreter.{AbstractOrMissingHandler, ILoop, IMain, JPri
 import scala.tools.nsc.interpreter.{NamedParam, SimpleReader, SplashLoop, SplashReader}
 import scala.tools.nsc.interpreter.StdReplTags.tagOfIMain
 import scala.tools.nsc.util.stringFromStream
-import scala.util.Properties.{javaVersion, javaVmName, versionString}
+import scala.util.Properties.{javaVersion, javaVmName, versionNumberString, versionString}
 
 /**
  *  A Spark-specific interactive shell.
@@ -43,10 +42,32 @@ class SparkILoop(in0: Option[BufferedReader], out: JPrintWriter)
   def this(in0: BufferedReader, out: JPrintWriter) = this(Some(in0), out)
   def this() = this(None, new JPrintWriter(Console.out, true))
 
+  /**
+   * TODO: Remove the following `override` when the support of Scala 2.11 is ended
+   * Scala 2.11 has a bug of finding imported types in class constructors, extends clause
+   * which is fixed in Scala 2.12 but never be back-ported into Scala 2.11.x.
+   * As a result, we copied the fixes into `SparkILoopInterpreter`. See SPARK-22393 for detail.
+   */
   override def createInterpreter(): Unit = {
-    intp = new SparkILoopInterpreter(settings, out)
+    if (isScala2_11) {
+      if (addedClasspath != "") {
+        settings.classpath append addedClasspath
+      }
+      // scalastyle:off classforname
+      // Have to use the default classloader to match the one used in
+      // `classOf[Settings]` and `classOf[JPrintWriter]`.
+      intp = Class.forName("org.apache.spark.repl.SparkILoopInterpreter")
+        .getDeclaredConstructor(Seq(classOf[Settings], classOf[JPrintWriter]): _*)
+        .newInstance(Seq(settings, out): _*)
+        .asInstanceOf[IMain]
+      // scalastyle:on classforname
+    } else {
+      super.createInterpreter()
+    }
   }
 
+  private val isScala2_11 = versionNumberString.startsWith("2.11")
+
   val initializationCommands: Seq[String] = Seq(
     """
     @transient val spark = if (org.apache.spark.repl.Main.sparkSession != null) {
@@ -124,6 +145,26 @@ class SparkILoop(in0: Option[BufferedReader], out: JPrintWriter)
     super.replay()
   }
 
+  /**
+   * TODO: Remove `runClosure` when the support of Scala 2.11 is ended
+   */
+  private def runClosure(body: => Boolean): Boolean = {
+    if (isScala2_11) {
+      // In Scala 2.11, there is a bug that interpret could set the current thread's
+      // context classloader, but fails to reset it to its previous state when returning
+      // from that method. This is fixed in SI-8521 https://github.com/scala/scala/pull/5657
+      // which is never back-ported into Scala 2.11.x. The following is a workaround fix.
+      val original = Thread.currentThread().getContextClassLoader
+      try {
+        body
+      } finally {
+        Thread.currentThread().setContextClassLoader(original)
+      }
+    } else {
+      body
+    }
+  }
+
   /**
    * The following code is mostly a copy of `process` implementation in `ILoop.scala` in Scala
    *
@@ -138,7 +179,7 @@ class SparkILoop(in0: Option[BufferedReader], out: JPrintWriter)
    * We should remove this duplication once Scala provides a way to load our custom initialization
    * code, and also customize the ordering of printing welcome message.
    */
-  override def process(settings: Settings): Boolean = savingContextLoader {
+  override def process(settings: Settings): Boolean = runClosure {
 
     def newReader = in0.fold(chooseReader(settings))(r => SimpleReader(r, out, interactive = true))