[livy] Don't start if spark-submit doesn't exist

apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/Main.scala

@@ -18,20 +18,24 @@
 
 package com.cloudera.hue.livy.server
 
+import java.io.IOException
+import java.nio.file.Path
 import javax.servlet.ServletContext
 
 import com.cloudera.hue.livy.server.batch.{BatchSessionProcessFactory, BatchSessionServlet, BatchManager, BatchSessionYarnFactory}
 import com.cloudera.hue.livy.server.interactive._
-import com.cloudera.hue.livy.{Utils, Logging, LivyConf, WebServer}
+import com.cloudera.hue.livy._
 import org.scalatra._
 import org.scalatra.servlet.ScalatraListener
+import org.slf4j.LoggerFactory
 
 object Main {
 
   val SESSION_KIND = "livy-server.session.kind"
   val THREAD_SESSION = "thread"
   val PROCESS_SESSION = "process"
   val YARN_SESSION = "yarn"
+  lazy val logger = LoggerFactory.getLogger(this.getClass)
 
   def main(args: Array[String]): Unit = {
     val livyConf = new LivyConf()
@@ -40,6 +44,9 @@ object Main {
     val host = livyConf.get("livy.server.host", "0.0.0.0")
     val port = livyConf.getInt("livy.server.port", 8998)
 
+    // Make sure the `spark-submit` program exists, otherwise much of livy won't work.
+    testSparkSubmit(livyConf)
+
     val server = new WebServer(host, port)
 
     server.context.setResourceBase("src/main/com/cloudera/hue/livy/server")
@@ -58,6 +65,55 @@ object Main {
       dispatch.Http.shutdown()
     }
   }
+
+  /**
+   * Test that the configured `spark-submit` executable exists.
+   *
+   * @param livyConf
+   */
+  private def testSparkSubmit(livyConf: LivyConf) = {
+    try {
+      // Ignore the version for now.
+      sparkSubmitVersion(livyConf) match {
+        case version @ "1.3.0" | "1.3.1" =>
+          logger.info(f"Using spark-submit version $version")
+        case version =>
+          logger.warn(f"Warning, livy has not been tested with spark-submit version $version")
+      }
+    } catch {
+      case e: IOException =>
+        System.err.println("Failed to run spark-submit executable: " + e.toString)
+        System.exit(1)
+    }
+  }
+
+  /**
+   * Return the version of the configured `spark-submit` version.
+   *
+   * @param livyConf
+   * @return the version
+   */
+  private def sparkSubmitVersion(livyConf: LivyConf): String = {
+    val sparkSubmit = livyConf.sparkSubmit()
+    val pb = new ProcessBuilder(sparkSubmit, "--version")
+    pb.redirectErrorStream(true)
+    pb.redirectInput(ProcessBuilder.Redirect.PIPE)
+
+    val process = new LineBufferedProcess(pb.start())
+    val exitCode = process.waitFor()
+    val output = process.inputIterator.mkString("\n")
+
+    if (exitCode != 1) {
+      throw new IOException(f"spark-submit had an unexpected exit [$exitCode]:\n$output]")
+    }
+
+    val regex = """version (.*)""".r.unanchored
+
+    output match {
+      case regex(version) => version
+      case _ => throw new IOException(f"Unable to determing spark-submit version:\n$output")
+    }
+  }
 }
 
 class ScalatraBootstrap extends LifeCycle with Logging {