Просмотр исходного кода

[livy] Fix AsyncResult exceptions being silently ignored

Erick Tryzelaar 10 лет назад
Родитель
Сommit
3b4e111479

+ 6 - 0
apps/spark/java/livy-core/pom.xml

@@ -51,6 +51,12 @@
             <version>${json4s.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-server</artifactId>
+            <version>${jetty.version}</version>
+        </dependency>
+
     </dependencies>
 
     <build>

+ 4 - 0
apps/spark/java/livy-core/src/main/scala/com/cloudera/hue/livy/Logging.scala

@@ -21,6 +21,10 @@ trait Logging {
     logger.warn(message.toString)
   }
 
+  def error(message: => Any, t: Throwable) = {
+    logger.error(message.toString, t)
+  }
+
   def error(message: => Any) = {
     logger.error(message.toString)
   }

+ 6 - 0
apps/spark/java/livy-repl/pom.xml

@@ -24,6 +24,12 @@
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-repl_2.10</artifactId>
             <version>${spark.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.eclipse.jetty</groupId>
+                    <artifactId>jetty-server</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>

+ 4 - 5
apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/Main.scala

@@ -2,6 +2,7 @@ package com.cloudera.hue.livy.repl
 
 import javax.servlet.ServletContext
 
+import com.cloudera.hue.livy.repl.spark.SparkSession
 import com.cloudera.hue.livy.{Logging, WebServer}
 import org.scalatra.LifeCycle
 import org.scalatra.servlet.ScalatraListener
@@ -25,15 +26,13 @@ object Main extends Logging {
 
 class ScalatraBootstrap extends LifeCycle {
 
-  //val system = ActorSystem()
-  val sparkInterpreter = new SparkInterpreter
+  val session = new SparkSession()
 
   override def init(context: ServletContext): Unit = {
-    context.mount(new WebApp(sparkInterpreter), "/*")
+    context.mount(new WebApp(session), "/*")
   }
 
   override def destroy(context: ServletContext): Unit = {
-    sparkInterpreter.close()
-    //system.shutdown()
+    session.close()
   }
 }

+ 15 - 0
apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/Session.scala

@@ -0,0 +1,15 @@
+package com.cloudera.hue.livy.repl
+
+import com.cloudera.hue.livy.ExecuteResponse
+
+import scala.concurrent.Future
+
+trait Session {
+  def statements: List[ExecuteResponse]
+
+  def statement(id: Int): Option[ExecuteResponse]
+
+  def execute(command: String): Future[ExecuteResponse]
+
+  def close(): Unit
+}

+ 18 - 12
apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/WebApp.scala

@@ -1,15 +1,17 @@
 package com.cloudera.hue.livy.repl
 
 import _root_.akka.util.Timeout
-import com.cloudera.hue.livy.ExecuteRequest
+import com.cloudera.hue.livy.{Logging, ExecuteRequest}
 import com.fasterxml.jackson.core.JsonParseException
-import org.json4s.{MappingException, DefaultFormats, Formats}
-import org.scalatra.json._
+import org.json4s.{DefaultFormats, Formats, MappingException}
+import org.scalatra.json.JacksonJsonSupport
 import org.scalatra._
 
 import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, Future}
 
-class WebApp(interpreter: SparkInterpreter) extends ScalatraServlet with FutureSupport with JacksonJsonSupport {
+object WebApp extends Logging {}
+
+class WebApp(session: Session) extends ScalatraServlet with FutureSupport with JacksonJsonSupport {
 
   override protected implicit def executor: ExecutionContextExecutor = ExecutionContext.global
   override protected implicit val jsonFormats: Formats = DefaultFormats
@@ -37,19 +39,20 @@ class WebApp(interpreter: SparkInterpreter) extends ScalatraServlet with FutureS
   }
 
   get("/statements") {
-    interpreter.statements
+    session.statements
   }
 
   post("/statements") {
     val req = parsedBody.extract[ExecuteRequest]
-    val statement = req.statement
-    new AsyncResult { val is = interpreter.execute(statement) }
+    val statement: String = req.statement
+    val rep = session.execute(statement)
+    new AsyncResult { val is = rep }
   }
 
   get("/statements/:statementId") {
     val statementId = params("statementId").toInt
 
-    interpreter.statement(statementId) match {
+    session.statement(statementId) match {
       case Some(statement) => statement
       case None => NotFound("Statement not found")
     }
@@ -58,7 +61,7 @@ class WebApp(interpreter: SparkInterpreter) extends ScalatraServlet with FutureS
   delete("/") {
     Future {
       state = ShuttingDown()
-      interpreter.close()
+      session.close()
       Thread.sleep(1000)
       System.exit(0)
     }
@@ -66,8 +69,11 @@ class WebApp(interpreter: SparkInterpreter) extends ScalatraServlet with FutureS
   }
 
   error {
-    case e: JsonParseException => halt(400, e.getMessage)
-    case e: MappingException => halt(400, e.getMessage)
-    case t => throw t
+    case e: JsonParseException => BadRequest(e.getMessage)
+    case e: MappingException => BadRequest(e.getMessage)
+    case e =>
+      WebApp.error("internal error", e)
+      InternalServerError(e.toString)
+      halt(500)
   }
 }

+ 4 - 59
apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/Interpreter.scala → apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/spark/ILoop.scala

@@ -1,72 +1,17 @@
-package com.cloudera.hue.livy.repl
+package com.cloudera.hue.livy.repl.spark
 
 import java.io._
-import java.util.concurrent.{BlockingQueue, SynchronousQueue}
+import java.util.concurrent.BlockingQueue
 
-import com.cloudera.hue.livy.ExecuteResponse
 import org.apache.spark.repl.SparkILoop
 
 import scala.annotation.tailrec
-import scala.collection.mutable
 import scala.concurrent._
-import scala.concurrent.duration.Duration
 import scala.tools.nsc.SparkHelper
 import scala.tools.nsc.interpreter.{Formatting, _}
 import scala.tools.nsc.util.ClassPath
 
-class SparkInterpreter {
-  private implicit def executor: ExecutionContext = ExecutionContext.global
-
-  private val inQueue = new SynchronousQueue[ILoop.Request]
-  private var executedStatements = 0
-  private var statements_ = new mutable.ArrayBuffer[ExecuteResponse]
-
-  org.apache.spark.repl.Main.interp = new ILoop(inQueue)
-
-  // Launch the real interpreter thread.
-  private val thread = new Thread {
-    override def run(): Unit = {
-      val args = Array("-usejavacp")
-      org.apache.spark.repl.Main.interp.process(args)
-    }
-  }
-  thread.start()
-
-  def statements: List[ExecuteResponse] = synchronized { statements_.toList }
-
-  def statement(id: Int): Option[ExecuteResponse] = synchronized {
-    if (id < statements_.length) {
-      Some(statements_(id))
-    } else {
-      None
-    }
-  }
-
-  def execute(statement: String): Future[ExecuteResponse] = {
-    executedStatements += 1
-
-    val promise = Promise[ILoop.ExecuteResponse]()
-    inQueue.put(ILoop.ExecuteRequest(statement, promise))
-
-    promise.future.map {
-      case rep =>
-        val executeResponse = ExecuteResponse(executedStatements - 1, List(statement), List(rep.output))
-        synchronized { statements_ += executeResponse }
-        executeResponse
-    }
-  }
-
-  def close(): Unit = {
-    val promise = Promise[ILoop.ShutdownResponse]()
-    inQueue.put(ILoop.ShutdownRequest(promise))
-
-    Await.result(promise.future, Duration.Inf)
-
-    thread.join()
-  }
-}
-
-private object ILoop {
+object ILoop {
   sealed trait Request
   case class ExecuteRequest(statement: String, promise: Promise[ExecuteResponse]) extends Request
   case class ShutdownRequest(promise: Promise[ShutdownResponse]) extends Request
@@ -76,7 +21,7 @@ private object ILoop {
 }
 
 // FIXME: The spark interpreter is written to own the event loop, so we need to invert it so we can inject our commands into it.
-private class ILoop(inQueue: BlockingQueue[ILoop.Request], outString: StringWriter = new StringWriter)
+class ILoop(inQueue: BlockingQueue[ILoop.Request], outString: StringWriter = new StringWriter)
   extends SparkILoop(
     // we don't actually use the reader, so pass in a null reader for now.
     new BufferedReader(new StringReader("")),

+ 64 - 0
apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/spark/SparkSession.scala

@@ -0,0 +1,64 @@
+package com.cloudera.hue.livy.repl.spark
+
+import java.util.concurrent.SynchronousQueue
+
+import com.cloudera.hue.livy.ExecuteResponse
+import com.cloudera.hue.livy.repl.Session
+
+import scala.collection.mutable
+import scala.concurrent.duration.Duration
+import scala.concurrent.{Await, ExecutionContext, Future, Promise}
+
+class SparkSession extends Session {
+  private implicit def executor: ExecutionContext = ExecutionContext.global
+
+  private[this] val inQueue = new SynchronousQueue[ILoop.Request]
+  private[this] var executedStatements = 0
+  private[this] var statements_ = new mutable.ArrayBuffer[ExecuteResponse]
+
+  org.apache.spark.repl.Main.interp = new ILoop(inQueue)
+
+  // Launch the real interpreter thread.
+  private[this] val thread = new Thread {
+    override def run(): Unit = {
+      val args = Array("-usejavacp")
+      org.apache.spark.repl.Main.interp.process(args)
+    }
+  }
+  thread.start()
+
+  override def statements: List[ExecuteResponse] = synchronized {
+    statements_.toList
+  }
+
+  override def statement(id: Int): Option[ExecuteResponse] = synchronized {
+    if (id < statements_.length) {
+      Some(statements_(id))
+    } else {
+      None
+    }
+  }
+
+  override def execute(statement: String): Future[ExecuteResponse] = {
+    executedStatements += 1
+
+    val promise = Promise[ILoop.ExecuteResponse]()
+    inQueue.put(ILoop.ExecuteRequest(statement, promise))
+
+    promise.future.map {
+      case rep =>
+        val executeResponse = ExecuteResponse(executedStatements - 1, List(statement), List(rep.output))
+        synchronized { statements_ += executeResponse }
+        executeResponse
+    }
+  }
+
+  override def close(): Unit = {
+    val promise = Promise[ILoop.ShutdownResponse]()
+    inQueue.put(ILoop.ShutdownRequest(promise))
+
+    Await.result(promise.future, Duration.Inf)
+
+    thread.join()
+  }
+}

+ 6 - 0
apps/spark/java/livy-server/pom.xml

@@ -68,6 +68,12 @@
             <groupId>com.cloudera.hue.livy</groupId>
             <artifactId>livy-yarn</artifactId>
             <version>${project.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.eclipse.jetty</groupId>
+                    <artifactId>jetty-server</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>

+ 15 - 18
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/WebApp.scala

@@ -1,5 +1,6 @@
 package com.cloudera.hue.livy.server
 
+import com.cloudera.hue.livy.Logging
 import com.cloudera.hue.livy.server.sessions.{SessionFailedToStart, Session}
 import com.fasterxml.jackson.core.JsonParseException
 import org.json4s.{DefaultFormats, Formats, MappingException}
@@ -9,7 +10,7 @@ import org.scalatra.json.JacksonJsonSupport
 import scala.concurrent._
 import scala.concurrent.duration._
 
-object WebApp {
+object WebApp extends Logging {
   case class CreateSessionRequest(lang: String)
   case class ExecuteStatementRequest(statement: String)
 }
@@ -21,7 +22,7 @@ class WebApp(sessionManager: SessionManager)
   with JacksonJsonSupport
   with UrlGeneratorSupport {
 
-  import com.cloudera.hue.livy.server.WebApp._
+  import WebApp._
 
   override protected implicit def executor: ExecutionContextExecutor = ExecutionContext.global
   override protected implicit def jsonFormats: Formats = DefaultFormats
@@ -55,9 +56,7 @@ class WebApp(sessionManager: SessionManager)
           headers = Map("Location" -> url(getSession, "sessionId" -> session.id.toString)))
     }
 
-    // FIXME: this is silently eating exceptions.
-    //new AsyncResult { val is = rep }
-    Await.result(rep, Duration.Inf)
+    new AsyncResult { val is = rep }
   }
 
   post("/sessions/:sessionId/stop") {
@@ -65,9 +64,7 @@ class WebApp(sessionManager: SessionManager)
       case Some(session) =>
         val future = session.stop()
 
-        // FIXME: this is silently eating exceptions.
-        //new AsyncResult() { val is = for { _ <- future } yield NoContent }
-        Await.result(future, Duration.Inf)
+        new AsyncResult() { val is = for { _ <- future } yield NoContent }
       case None => NotFound("Session not found")
     }
   }
@@ -80,8 +77,7 @@ class WebApp(sessionManager: SessionManager)
         } yield Accepted()
 
         // FIXME: this is silently eating exceptions.
-        //new AsyncResult() { val is = for { _ <- future } yield NoContent }
-        Await.result(future, Duration.Inf)
+        new AsyncResult() { val is = for { _ <- future } yield NoContent }
       case None => NotFound("Session not found")
     }
   }
@@ -92,8 +88,7 @@ class WebApp(sessionManager: SessionManager)
     } yield Accepted()
 
     // FIXME: this is silently eating exceptions.
-    //new AsyncResult() { val is = for { _ <- future } yield NoContent }
-    Await.result(future, Duration.Inf)
+    new AsyncResult() { val is = for { _ <- future } yield NoContent }
   }
 
   get("/sessions/:sessionId/statements") {
@@ -130,13 +125,15 @@ class WebApp(sessionManager: SessionManager)
     }
   }
 
-
   error {
-    case e: JsonParseException => halt(400, e.getMessage)
-    case e: MappingException => halt(400, e.getMessage)
-    case e: SessionFailedToStart => halt(500, e.getMessage)
-    case e: dispatch.StatusCode => halt(e.code, e.getMessage)
-    case t => throw t
+    case e: JsonParseException => BadRequest(e.getMessage)
+    case e: MappingException => BadRequest(e.getMessage)
+    case e: SessionFailedToStart => InternalServerError(e.getMessage)
+    case e: dispatch.StatusCode => ActionResult(ResponseStatus(e.code), e.getMessage, Map.empty)
+    case e =>
+      WebApp.error("internal error", e)
+      InternalServerError(e.toString)
+      halt(500)
   }
 
   private def formatSession(session: Session) = {