Explorar el Código

[livy] Error out the session if the repl errors out

Potentially related to #235.
Erick Tryzelaar hace 10 años
padre
commit
a1ae994

+ 37 - 13
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/interactive/InteractiveWebSession.scala

@@ -25,7 +25,7 @@ import com.cloudera.hue.livy._
 import com.cloudera.hue.livy.msgs.ExecuteRequest
 import com.cloudera.hue.livy.sessions._
 import dispatch._
-import org.json4s.JsonAST.JNull
+import org.json4s.JsonAST.{JString, JNull}
 import org.json4s.jackson.Serialization.write
 import org.json4s.{DefaultFormats, Formats, JValue}
 
@@ -77,14 +77,10 @@ abstract class InteractiveWebSession(val id: Int, createInteractiveRequest: Crea
       val req = (svc / "execute").setContentType("application/json", "UTF-8") << write(content)
 
       val future = Http(req OK as.json4s.Json).map { case resp: JValue =>
-        resp \ "result" match {
-          case JNull =>
-            // The result isn't ready yet. Loop until it is.
-            val id = (resp \ "id").extract[Int]
-            waitForStatement(id)
-          case result =>
-            transition(Idle())
-            result
+        parseResponse(resp).getOrElse {
+          // The result isn't ready yet. Loop until it is.
+          val id = (resp \ "id").extract[Int]
+          waitForStatement(id)
         }
       }
 
@@ -102,13 +98,41 @@ abstract class InteractiveWebSession(val id: Int, createInteractiveRequest: Crea
     val req = (svc / "history" / id).setContentType("application/json", "UTF-8")
     val resp = Await.result(Http(req OK as.json4s.Json), Duration.Inf)
 
-    resp \ "result" match {
-      case JNull =>
+    parseResponse(resp) match {
+      case Some(result) => result
+      case None =>
         Thread.sleep(1000)
         waitForStatement(id)
+    }
+  }
+
+  private def parseResponse(response: JValue): Option[JValue] = {
+    response \ "result" match {
+      case JNull => None
       case result =>
-        transition(Idle())
-        result
+        // If the response errored out, it's possible it took down the interpreter. Check if
+        // it's still running.
+        result \ "status" match {
+          case JString("error") =>
+            if (replErroredOut()) {
+              transition(Error())
+            } else {
+              transition(Idle())
+            }
+          case _ => transition(Idle())
+        }
+
+        Some(result)
+    }
+  }
+
+  private def replErroredOut() = {
+    val req = svc.setContentType("application/json", "UTF-8")
+    val response = Await.result(Http(req OK as.json4s.Json), Duration.Inf)
+
+    response \ "state" match {
+      case JString("error") => true
+      case _ => false
     }
   }
 

+ 16 - 8
apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/interactive/BaseSessionSpec.scala

@@ -21,7 +21,7 @@ package com.cloudera.hue.livy.server.interactive
 import java.util.concurrent.TimeUnit
 
 import com.cloudera.hue.livy.msgs.ExecuteRequest
-import com.cloudera.hue.livy.sessions.{Idle, Starting}
+import com.cloudera.hue.livy.sessions.{Error, Idle, Starting}
 import org.json4s.{DefaultFormats, Extraction}
 import org.scalatest.{BeforeAndAfter, FunSpec, Matchers}
 
@@ -63,7 +63,7 @@ abstract class BaseSessionSpec extends FunSpec with Matchers with BeforeAndAfter
         "status" -> "ok",
         "execution_count" -> 0,
         "data" -> Map(
-          "text/plain" -> "res0: Int = 3"
+          "text/plain" -> "3"
         )
       ))
 
@@ -77,15 +77,23 @@ abstract class BaseSessionSpec extends FunSpec with Matchers with BeforeAndAfter
       val expectedResult = Extraction.decompose(Map(
         "status" -> "error",
         "execution_count" -> 0,
-        "ename" -> "Error",
-        "evalue" ->
-          """<console>:8: error: not found: value x
-            |              x
-            |              ^""".stripMargin,
-        "traceback" -> List()
+        "ename" -> "NameError",
+        "evalue" -> "name 'x' is not defined",
+        "traceback" -> List(
+          "Traceback (most recent call last):\n",
+          "NameError: name 'x' is not defined\n"
+        )
       ))
 
       result should equal (expectedResult)
+      session.state should equal (Idle())
+    }
+
+    it("should error out the session if the interpreter dies") {
+      session.waitForStateChange(Starting(), Duration(30, TimeUnit.SECONDS))
+      val stmt = session.executeStatement(ExecuteRequest("import os; os._exit(1)"))
+      val result = Await.result(stmt.output(), Duration.Inf)
+      session.state should equal (Error())
     }
   }
 }

+ 2 - 2
apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/interactive/InteractiveSessionProcessSpec.scala

@@ -19,7 +19,7 @@
 package com.cloudera.hue.livy.server.interactive
 
 import com.cloudera.hue.livy.LivyConf
-import com.cloudera.hue.livy.sessions.Spark
+import com.cloudera.hue.livy.sessions.PySpark
 import org.scalatest.{BeforeAndAfter, FunSpecLike, Matchers}
 
 class InteractiveSessionProcessSpec extends BaseSessionSpec with FunSpecLike with Matchers with BeforeAndAfter {
@@ -27,5 +27,5 @@ class InteractiveSessionProcessSpec extends BaseSessionSpec with FunSpecLike wit
   val livyConf = new LivyConf()
   livyConf.set("livy.repl.driverClassPath", sys.props("java.class.path"))
 
-  def createSession() = InteractiveSessionProcess.create(livyConf, 0, CreateInteractiveRequest(kind = Spark()))
+  def createSession() = InteractiveSessionProcess.create(livyConf, 0, CreateInteractiveRequest(kind = PySpark()))
 }