LIVY-246 Support multiple Spark home in runtime

Author: Agarwal

server/src/main/scala/com/cloudera/livy/LivyConf.scala

@@ -246,14 +246,20 @@ class LivyConf(loadDefaults: Boolean) extends ClientConf[LivyConf](null) {
   def sparkDeployMode(): Option[String] = Option(get(LIVY_SPARK_DEPLOY_MODE)).filterNot(_.isEmpty)
 
   /** Return the location of the spark home directory */
-  def sparkHome(): Option[String] = Option(get(SPARK_HOME)).orElse(sys.env.get("SPARK_HOME"))
+  def sparkHome(version: Option[String] = None): Option[String] = {
+    version.map {version => Option(get(s"livy.server.spark-home_$version"))
+      .orElse(throw new IllegalArgumentException(
+        s"Spark version: $version is not supported"))
+    }.getOrElse(Option(get(s"livy.server.spark-home")).orElse(sys.env.get("SPARK_HOME")))
+  }
 
   /** Return the spark master Livy sessions should use. */
   def sparkMaster(): String = get(LIVY_SPARK_MASTER)
 
   /** Return the path to the spark-submit executable. */
-  def sparkSubmit(): String = {
-    sparkHome().map { _ + File.separator + "bin" + File.separator + "spark-submit" }.get
+  def sparkSubmit(version: Option[String] = None): String = {
+    sparkHome(version).map { _ + File.separator + "bin" + File.separator +
+      "spark-submit" }.get
   }
 
   /** Return the list of superusers. */

server/src/main/scala/com/cloudera/livy/server/batch/BatchSession.scala

@@ -18,6 +18,7 @@
 
 package com.cloudera.livy.server.batch
 
+import java.io.File
 import java.lang.ProcessBuilder.Redirect
 
 import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
@@ -62,7 +63,11 @@ object BatchSession {
           request.conf, request.jars, request.files, request.archives, request.pyFiles, livyConf))
       require(request.file != null, "File is required.")
 
-      val builder = new SparkProcessBuilder(livyConf)
+      val builder = new SparkProcessBuilder(livyConf, request.sparkVersion)
+      request.sparkVersion.map({ value =>
+        builder.env("SPARK_CONF_DIR", livyConf.sparkHome(request.sparkVersion)
+          + File.separator + "conf")
+      })
       builder.conf(conf)
 
       proxyUser.foreach(builder.proxyUser)

server/src/main/scala/com/cloudera/livy/server/batch/CreateBatchRequest.scala

@@ -36,5 +36,6 @@ class CreateBatchRequest {
   var queue: Option[String] = None
   var name: Option[String] = None
   var conf: Map[String, String] = Map()
+  var sparkVersion: Option[String] = None
 
 }

server/src/main/scala/com/cloudera/livy/utils/SparkProcessBuilder.scala

@@ -25,9 +25,9 @@ import scala.collection.mutable.ArrayBuffer
 import com.cloudera.livy.{LivyConf, Logging}
 import com.cloudera.livy.util.LineBufferedProcess
 
-class SparkProcessBuilder(livyConf: LivyConf) extends Logging {
+class SparkProcessBuilder(livyConf: LivyConf, version: Option[String]) extends Logging  {
 
-  private[this] var _executable: String = livyConf.sparkSubmit()
+  private[this] var _executable: String = livyConf.sparkSubmit(version)
   private[this] var _master: Option[String] = None
   private[this] var _deployMode: Option[String] = None
   private[this] var _className: Option[String] = None