Browse Source

[livy] Reorganize the code in prep for batch jobs

Erick Tryzelaar 10 years ago
parent
commit
6d27742740
23 changed files with 199 additions and 223 deletions
  1. 10 0
      apps/spark/java/livy-core/src/main/scala/com/cloudera/hue/livy/sessions/Kind.scala
  2. 31 0
      apps/spark/java/livy-core/src/main/scala/com/cloudera/hue/livy/sessions/State.scala
  3. 3 2
      apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/Main.scala
  4. 2 24
      apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/Session.scala
  5. 9 8
      apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/WebApp.scala
  6. 18 18
      apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/python/PythonSession.scala
  7. 9 8
      apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/scala/SparkSession.scala
  8. 37 0
      apps/spark/java/livy-repl/src/test/scala/com/cloudera/hue/livy/repl/BaseSessionSpec.scala
  9. 3 25
      apps/spark/java/livy-repl/src/test/scala/com/cloudera/hue/livy/repl/PythonSessionSpec.scala
  10. 3 25
      apps/spark/java/livy-repl/src/test/scala/com/cloudera/hue/livy/repl/SparkSessionSpec.scala
  11. 6 6
      apps/spark/java/livy-repl/src/test/scala/com/cloudera/hue/livy/repl/WebAppSpec.scala
  12. 1 1
      apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/Main.scala
  13. 4 3
      apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/ProcessSession.scala
  14. 1 39
      apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/Session.scala
  15. 5 4
      apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/SessionFactory.scala
  16. 2 1
      apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/SessionManager.scala
  17. 24 23
      apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/SessionServlet.scala
  18. 6 18
      apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/ThreadSession.scala
  19. 7 3
      apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/WebSession.scala
  20. 5 4
      apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/YarnSession.scala
  21. 6 5
      apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/BaseSessionSpec.scala
  22. 4 4
      apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/ProcessSessionSpec.scala
  23. 3 2
      apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/ThreadSessionSpec.scala

+ 10 - 0
apps/spark/java/livy-core/src/main/scala/com/cloudera/hue/livy/sessions/Kind.scala

@@ -0,0 +1,10 @@
+package com.cloudera.hue.livy.sessions
+
+sealed trait Kind
+case class Spark() extends Kind {
+  override def toString = "spark"
+}
+
+case class PySpark() extends Kind {
+  override def toString = "pyspark"
+}

+ 31 - 0
apps/spark/java/livy-core/src/main/scala/com/cloudera/hue/livy/sessions/State.scala

@@ -0,0 +1,31 @@
+package com.cloudera.hue.livy.sessions
+
+sealed trait State
+
+case class NotStarted() extends State {
+  override def toString = "not_started"
+}
+
+case class Starting() extends State {
+  override def toString = "starting"
+}
+
+case class Idle() extends State {
+  override def toString = "idle"
+}
+
+case class Busy() extends State {
+  override def toString = "busy"
+}
+
+case class Error() extends State {
+  override def toString = "error"
+}
+
+case class ShuttingDown() extends State {
+  override def toString = "shutting_down"
+}
+
+case class Dead() extends State {
+  override def toString = "dead"
+}

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

@@ -5,7 +5,8 @@ import javax.servlet.ServletContext
 
 
 import com.cloudera.hue.livy.repl.python.PythonSession
 import com.cloudera.hue.livy.repl.python.PythonSession
 import com.cloudera.hue.livy.repl.scala.SparkSession
 import com.cloudera.hue.livy.repl.scala.SparkSession
-import com.cloudera.hue.livy.{Utils, Logging, WebServer}
+import com.cloudera.hue.livy.sessions.Starting
+import com.cloudera.hue.livy.{Logging, WebServer}
 import dispatch._
 import dispatch._
 import org.json4s.jackson.Serialization.write
 import org.json4s.jackson.Serialization.write
 import org.json4s.{DefaultFormats, Formats}
 import org.json4s.{DefaultFormats, Formats}
@@ -106,7 +107,7 @@ class ScalatraBootstrap extends LifeCycle with Logging {
     info(s"Notifying $callbackUrl that we're up")
     info(s"Notifying $callbackUrl that we're up")
 
 
     Future {
     Future {
-      session.waitForStateChange(Session.Starting(), Duration(10, TimeUnit.SECONDS))
+      session.waitForStateChange(Starting(), Duration(10, TimeUnit.SECONDS))
 
 
       // Wait for our url to be discovered.
       // Wait for our url to be discovered.
       val replUrl = waitForReplUrl()
       val replUrl = waitForReplUrl()

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

@@ -1,35 +1,13 @@
 package com.cloudera.hue.livy.repl
 package com.cloudera.hue.livy.repl
 
 
 import com.cloudera.hue.livy.Utils
 import com.cloudera.hue.livy.Utils
+import com.cloudera.hue.livy.sessions.{Kind, State}
 import org.json4s.JValue
 import org.json4s.JValue
 
 
-import _root_.scala.annotation.tailrec
 import _root_.scala.concurrent.duration.Duration
 import _root_.scala.concurrent.duration.Duration
-import _root_.scala.concurrent.{TimeoutException, Future}
-
-object Session {
-  sealed trait State
-  case class NotStarted() extends State
-  case class Starting() extends State
-  case class Idle() extends State
-  case class Busy() extends State
-  case class Error() extends State
-  case class ShuttingDown() extends State
-  case class ShutDown() extends State
-
-  sealed trait Kind
-  case class Spark() extends Kind {
-    override def toString = "spark"
-  }
-
-  case class PySpark() extends Kind {
-    override def toString = "pyspark"
-  }
-}
+import _root_.scala.concurrent.{Future, TimeoutException}
 
 
 trait Session {
 trait Session {
-  import Session._
-
   def kind: Kind
   def kind: Kind
 
 
   def state: State
   def state: State

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

@@ -2,6 +2,7 @@ package com.cloudera.hue.livy.repl
 
 
 import com.cloudera.hue.livy.Logging
 import com.cloudera.hue.livy.Logging
 import com.cloudera.hue.livy.msgs.ExecuteRequest
 import com.cloudera.hue.livy.msgs.ExecuteRequest
+import com.cloudera.hue.livy.sessions._
 import com.fasterxml.jackson.core.JsonParseException
 import com.fasterxml.jackson.core.JsonParseException
 import org.json4s.{DefaultFormats, MappingException}
 import org.json4s.{DefaultFormats, MappingException}
 import org.scalatra._
 import org.scalatra._
@@ -20,20 +21,20 @@ class WebApp(session: Session) extends ScalatraServlet with FutureSupport with J
     contentType = formats("json")
     contentType = formats("json")
 
 
     session.state match {
     session.state match {
-      case Session.ShuttingDown() => halt(500, "Shutting down")
+      case ShuttingDown() => halt(500, "Shutting down")
       case _ => {}
       case _ => {}
     }
     }
   }
   }
 
 
   get("/") {
   get("/") {
     val state = session.state match {
     val state = session.state match {
-      case Session.NotStarted() => "not_started"
-      case Session.Starting() => "starting"
-      case Session.Idle() => "idle"
-      case Session.Busy() => "busy"
-      case Session.Error() => "error"
-      case Session.ShuttingDown() => "shutting_down"
-      case Session.ShutDown() => "shut_down"
+      case NotStarted() => "not_started"
+      case Starting() => "starting"
+      case Idle() => "idle"
+      case Busy() => "busy"
+      case Error() => "error"
+      case ShuttingDown() => "shutting_down"
+      case Dead() => "dead"
     }
     }
     Map("state" -> state)
     Map("state" -> state)
   }
   }

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

@@ -3,11 +3,11 @@ package com.cloudera.hue.livy.repl.python
 import java.io._
 import java.io._
 import java.lang.ProcessBuilder.Redirect
 import java.lang.ProcessBuilder.Redirect
 import java.nio.file.Files
 import java.nio.file.Files
-import java.util.concurrent.{TimeUnit, SynchronousQueue}
+import java.util.concurrent.{SynchronousQueue, TimeUnit}
 
 
-
-import com.cloudera.hue.livy.{Logging, Utils}
 import com.cloudera.hue.livy.repl.Session
 import com.cloudera.hue.livy.repl.Session
+import com.cloudera.hue.livy.sessions._
+import com.cloudera.hue.livy.{Logging, Utils}
 import org.apache.spark.SparkContext
 import org.apache.spark.SparkContext
 import org.json4s.jackson.JsonMethods._
 import org.json4s.jackson.JsonMethods._
 import org.json4s.jackson.Serialization.write
 import org.json4s.jackson.Serialization.write
@@ -17,8 +17,8 @@ import py4j.GatewayServer
 import scala.annotation.tailrec
 import scala.annotation.tailrec
 import scala.collection.JavaConversions._
 import scala.collection.JavaConversions._
 import scala.collection.mutable.ArrayBuffer
 import scala.collection.mutable.ArrayBuffer
-import scala.concurrent.duration.Duration
 import scala.concurrent._
 import scala.concurrent._
+import scala.concurrent.duration.Duration
 
 
 object PythonSession {
 object PythonSession {
   def createPython(): Session = {
   def createPython(): Session = {
@@ -113,7 +113,7 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
   private val stdout = new BufferedReader(new InputStreamReader(process.getInputStream), 1)
   private val stdout = new BufferedReader(new InputStreamReader(process.getInputStream), 1)
 
 
   private var _history = ArrayBuffer[JValue]()
   private var _history = ArrayBuffer[JValue]()
-  private var _state: Session.State = Session.Starting()
+  private var _state: State = Starting()
 
 
   private val queue = new SynchronousQueue[Request]
   private val queue = new SynchronousQueue[Request]
 
 
@@ -121,7 +121,7 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
     override def run() = {
     override def run() = {
       waitUntilReady()
       waitUntilReady()
 
 
-      _state = Session.Idle()
+      _state = Idle()
 
 
       loop()
       loop()
     }
     }
@@ -145,14 +145,14 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
     @tailrec
     @tailrec
     def loop(): Unit = {
     def loop(): Unit = {
       (_state, queue.take()) match {
       (_state, queue.take()) match {
-        case (Session.Error(), ExecuteRequest(code, promise)) =>
+        case (Error(), ExecuteRequest(code, promise)) =>
           promise.failure(new Exception("session has been terminated"))
           promise.failure(new Exception("session has been terminated"))
           loop()
           loop()
 
 
         case (state, ExecuteRequest(code, promise)) =>
         case (state, ExecuteRequest(code, promise)) =>
-          require(state == Session.Idle())
+          require(state == Idle())
 
 
-          _state = Session.Busy()
+          _state = Busy()
 
 
           sendRequest(Map("msg_type" -> "execute_request", "content" -> Map("code" -> code))) match {
           sendRequest(Map("msg_type" -> "execute_request", "content" -> Map("code" -> code))) match {
             case Some(rep) =>
             case Some(rep) =>
@@ -161,19 +161,19 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
               val content: JValue = rep \ "content"
               val content: JValue = rep \ "content"
               _history += content
               _history += content
 
 
-              _state = Session.Idle()
+              _state = Idle()
 
 
               promise.success(content)
               promise.success(content)
               loop()
               loop()
             case None =>
             case None =>
-              _state = Session.Error()
+              _state = Error()
               promise.failure(new Exception("session has been terminated"))
               promise.failure(new Exception("session has been terminated"))
           }
           }
 
 
         case (_, ShutdownRequest(promise)) =>
         case (_, ShutdownRequest(promise)) =>
-          require(state == Session.Idle() || state == Session.Error())
+          require(state == Idle() || state == Error())
 
 
-          _state = Session.ShuttingDown()
+          _state = ShuttingDown()
 
 
           try {
           try {
             sendRequest(Map("msg_type" -> "shutdown_request", "content" -> ())) match {
             sendRequest(Map("msg_type" -> "shutdown_request", "content" -> ())) match {
@@ -188,7 +188,7 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
             try {
             try {
               process.destroy()
               process.destroy()
             } finally {
             } finally {
-              _state = Session.ShutDown()
+              _state = Dead()
               promise.success(())
               promise.success(())
             }
             }
           }
           }
@@ -198,7 +198,7 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
 
 
   thread.start()
   thread.start()
 
 
-  override def kind = Session.PySpark()
+  override def kind = PySpark()
 
 
   override def state = _state
   override def state = _state
 
 
@@ -220,10 +220,10 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
 
 
   override def close(): Unit = synchronized {
   override def close(): Unit = synchronized {
     _state match {
     _state match {
-      case Session.ShutDown() =>
-      case Session.ShuttingDown() =>
+      case Dead() =>
+      case ShuttingDown() =>
         // Another thread must be tearing down the process.
         // Another thread must be tearing down the process.
-        waitForStateChange(Session.ShuttingDown(), Duration(10, TimeUnit.SECONDS))
+        waitForStateChange(ShuttingDown(), Duration(10, TimeUnit.SECONDS))
       case _ =>
       case _ =>
         val promise = Promise[Unit]()
         val promise = Promise[Unit]()
         queue.put(ShutdownRequest(promise))
         queue.put(ShutdownRequest(promise))

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

@@ -2,6 +2,7 @@ package com.cloudera.hue.livy.repl.scala
 
 
 import com.cloudera.hue.livy.repl.Session
 import com.cloudera.hue.livy.repl.Session
 import com.cloudera.hue.livy.repl.scala.interpreter._
 import com.cloudera.hue.livy.repl.scala.interpreter._
+import com.cloudera.hue.livy.sessions._
 import org.json4s.jackson.JsonMethods._
 import org.json4s.jackson.JsonMethods._
 import org.json4s.jackson.Serialization.write
 import org.json4s.jackson.Serialization.write
 import org.json4s.{JValue, _}
 import org.json4s.{JValue, _}
@@ -22,15 +23,15 @@ private class SparkSession extends Session {
   private val interpreter = new Interpreter()
   private val interpreter = new Interpreter()
   interpreter.start()
   interpreter.start()
 
 
-  override def kind: Session.Kind = Session.Spark()
+  override def kind: Kind = Spark()
 
 
-  override def state: Session.State = interpreter.state match {
-    case Interpreter.NotStarted() => Session.NotStarted()
-    case Interpreter.Starting() => Session.Starting()
-    case Interpreter.Idle() => Session.Idle()
-    case Interpreter.Busy() => Session.Busy()
-    case Interpreter.ShuttingDown() => Session.ShuttingDown()
-    case Interpreter.ShutDown() => Session.ShutDown()
+  override def state: State = interpreter.state match {
+    case Interpreter.NotStarted() => NotStarted()
+    case Interpreter.Starting() => Starting()
+    case Interpreter.Idle() => Idle()
+    case Interpreter.Busy() => Busy()
+    case Interpreter.ShuttingDown() => ShuttingDown()
+    case Interpreter.ShutDown() => Dead()
   }
   }
 
 
   override def history(): Seq[JValue] = _history
   override def history(): Seq[JValue] = _history

+ 37 - 0
apps/spark/java/livy-repl/src/test/scala/com/cloudera/hue/livy/repl/BaseSessionSpec.scala

@@ -0,0 +1,37 @@
+package com.cloudera.hue.livy.repl
+
+import java.util.concurrent.TimeUnit
+
+import com.cloudera.hue.livy.sessions.{Idle, Starting}
+import org.json4s.DefaultFormats
+import org.scalatest.{Matchers, FunSpec, BeforeAndAfter}
+
+import _root_.scala.concurrent.duration.Duration
+
+abstract class BaseSessionSpec extends FunSpec with Matchers with BeforeAndAfter {
+
+  implicit val formats = DefaultFormats
+
+  def createSession(): Session
+
+  var session: Session = null
+
+  before {
+    session = createSession()
+  }
+
+  after {
+    session.close()
+  }
+
+  describe("A session") {
+    it("should start in the starting or idle state") {
+      session.state should (equal (Starting()) or equal (Idle()))
+    }
+
+    it("should eventually become the idle state") {
+      session.waitForStateChange(Starting(), Duration(10, TimeUnit.SECONDS))
+      session.state should equal (Idle())
+    }
+  }
+}

+ 3 - 25
apps/spark/java/livy-repl/src/test/scala/com/cloudera/hue/livy/repl/PythonSessionSpec.scala

@@ -1,39 +1,17 @@
 package com.cloudera.hue.livy.repl
 package com.cloudera.hue.livy.repl
 
 
-import java.util.concurrent.TimeUnit
-
 import com.cloudera.hue.livy.repl.python.PythonSession
 import com.cloudera.hue.livy.repl.python.PythonSession
+import org.json4s.Extraction
 import org.json4s.JsonAST.JValue
 import org.json4s.JsonAST.JValue
-import org.json4s.{Extraction, DefaultFormats}
-import org.scalatest.{Matchers, BeforeAndAfter, FunSpec}
 
 
 import _root_.scala.concurrent.Await
 import _root_.scala.concurrent.Await
 import _root_.scala.concurrent.duration.Duration
 import _root_.scala.concurrent.duration.Duration
 
 
-class PythonSessionSpec extends FunSpec with Matchers with BeforeAndAfter {
-
-  implicit val formats = DefaultFormats
-
-  var session: Session = null
-
-  before {
-    session = PythonSession.createPython()
-  }
+class PythonSessionSpec extends BaseSessionSpec {
 
 
-  after {
-    session.close()
-  }
+  override def createSession() = PythonSession.createPySpark()
 
 
   describe("A python session") {
   describe("A python session") {
-    it("should start in the starting or idle state") {
-      session.state should (equal (Session.Starting()) or equal (Session.Idle()))
-    }
-
-    it("should eventually become the idle state") {
-      session.waitForStateChange(Session.Starting(), Duration(10, TimeUnit.SECONDS))
-      session.state should equal (Session.Idle())
-    }
-
     it("should execute `1 + 2` == 3") {
     it("should execute `1 + 2` == 3") {
       val result = Await.result(session.execute("1 + 2"), Duration.Inf)
       val result = Await.result(session.execute("1 + 2"), Duration.Inf)
       val expectedResult = Extraction.decompose(Map(
       val expectedResult = Extraction.decompose(Map(

+ 3 - 25
apps/spark/java/livy-repl/src/test/scala/com/cloudera/hue/livy/repl/SparkSessionSpec.scala

@@ -1,39 +1,17 @@
 package com.cloudera.hue.livy.repl
 package com.cloudera.hue.livy.repl
 
 
-import java.util.concurrent.TimeUnit
-
 import com.cloudera.hue.livy.repl.scala.SparkSession
 import com.cloudera.hue.livy.repl.scala.SparkSession
+import org.json4s.Extraction
 import org.json4s.JsonAST.JValue
 import org.json4s.JsonAST.JValue
-import org.json4s.{DefaultFormats, Extraction}
-import org.scalatest.{BeforeAndAfter, FunSpec, Matchers}
 
 
 import _root_.scala.concurrent.Await
 import _root_.scala.concurrent.Await
 import _root_.scala.concurrent.duration.Duration
 import _root_.scala.concurrent.duration.Duration
 
 
-class SparkSessionSpec extends FunSpec with Matchers with BeforeAndAfter {
-
-  implicit val formats = DefaultFormats
-
-  var session: Session = null
-
-  before {
-    session = SparkSession.create()
-  }
+class SparkSessionSpec extends BaseSessionSpec {
 
 
-  after {
-    session.close()
-  }
+  override def createSession() = SparkSession.create()
 
 
   describe("A spark session") {
   describe("A spark session") {
-    it("should start in the starting or idle state") {
-      session.state should (equal (Session.Starting()) or equal (Session.Idle()))
-    }
-
-    it("should eventually become the idle state") {
-      session.waitForStateChange(Session.Starting(), Duration(10, TimeUnit.SECONDS))
-      session.state should equal (Session.Idle())
-    }
-
     it("should execute `1 + 2` == 3") {
     it("should execute `1 + 2` == 3") {
       val result = Await.result(session.execute("1 + 2"), Duration.Inf)
       val result = Await.result(session.execute("1 + 2"), Duration.Inf)
       val expectedResult = Extraction.decompose(Map(
       val expectedResult = Extraction.decompose(Map(

+ 6 - 6
apps/spark/java/livy-repl/src/test/scala/com/cloudera/hue/livy/repl/WebAppSpec.scala

@@ -1,6 +1,6 @@
 package com.cloudera.hue.livy.repl
 package com.cloudera.hue.livy.repl
 
 
-import com.cloudera.hue.livy.repl.Session.{Kind, State}
+import com.cloudera.hue.livy.sessions._
 import org.json4s.JsonAST.{JArray, JString}
 import org.json4s.JsonAST.{JArray, JString}
 import org.json4s.JsonDSL._
 import org.json4s.JsonDSL._
 import org.json4s.jackson.JsonMethods._
 import org.json4s.jackson.JsonMethods._
@@ -14,10 +14,10 @@ class WebAppSpec extends ScalatraSuite with FunSpecLike with BeforeAndAfter {
   implicit val formats = DefaultFormats
   implicit val formats = DefaultFormats
 
 
   class MockSession extends Session {
   class MockSession extends Session {
-    var _state: State = Session.Idle()
+    var _state: State = Idle()
     var _history = List[JValue]()
     var _history = List[JValue]()
 
 
-    override def kind: Kind = Session.Spark()
+    override def kind: Kind = Spark()
 
 
     override def state = _state
     override def state = _state
 
 
@@ -27,7 +27,7 @@ class WebAppSpec extends ScalatraSuite with FunSpecLike with BeforeAndAfter {
     }
     }
 
 
     override def close(): Unit = {
     override def close(): Unit = {
-      _state = Session.ShuttingDown()
+      _state = Dead()
     }
     }
 
 
     override def history(): Seq[JValue] = _history
     override def history(): Seq[JValue] = _history
@@ -49,7 +49,7 @@ class WebAppSpec extends ScalatraSuite with FunSpecLike with BeforeAndAfter {
         parsedBody \ "state" should equal (JString("idle"))
         parsedBody \ "state" should equal (JString("idle"))
       }
       }
 
 
-      session._state = Session.Busy()
+      session._state = Busy()
 
 
       get("/") {
       get("/") {
         status should equal (200)
         status should equal (200)
@@ -80,7 +80,7 @@ class WebAppSpec extends ScalatraSuite with FunSpecLike with BeforeAndAfter {
   }
   }
 
 
   after {
   after {
-    session._state = Session.Idle()
+    session._state = Idle()
     session._history = List()
     session._history = List()
   }
   }
 }
 }

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

@@ -63,7 +63,7 @@ class ScalatraBootstrap extends LifeCycle with Logging {
 
 
     sessionManager = new SessionManager(sessionFactory)
     sessionManager = new SessionManager(sessionFactory)
 
 
-    context.mount(new WebApp(sessionManager), "/*")
+    context.mount(new SessionServlet(sessionManager), "/sessions/*")
   }
   }
 
 
   override def destroy(context: ServletContext): Unit = {
   override def destroy(context: ServletContext): Unit = {

+ 4 - 3
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/ProcessSession.scala

@@ -3,6 +3,7 @@ package com.cloudera.hue.livy.server.sessions
 import java.lang.ProcessBuilder.Redirect
 import java.lang.ProcessBuilder.Redirect
 import java.net.URL
 import java.net.URL
 
 
+import com.cloudera.hue.livy.sessions.Kind
 import com.cloudera.hue.livy.spark.SparkProcessBuilder
 import com.cloudera.hue.livy.spark.SparkProcessBuilder
 import com.cloudera.hue.livy.{LivyConf, Logging, Utils}
 import com.cloudera.hue.livy.{LivyConf, Logging, Utils}
 
 
@@ -18,13 +19,13 @@ object ProcessSession extends Logging {
   val CONF_LIVY_REPL_CALLBACK_URL = "livy.repl.callback-url"
   val CONF_LIVY_REPL_CALLBACK_URL = "livy.repl.callback-url"
   val CONF_LIVY_REPL_DRIVER_CLASS_PATH = "livy.repl.driverClassPath"
   val CONF_LIVY_REPL_DRIVER_CLASS_PATH = "livy.repl.driverClassPath"
 
 
-  def create(livyConf: LivyConf, id: String, kind: Session.Kind, proxyUser: Option[String] = None): Session = {
+  def create(livyConf: LivyConf, id: String, kind: Kind, proxyUser: Option[String] = None): Session = {
     val process = startProcess(livyConf, id, kind, proxyUser)
     val process = startProcess(livyConf, id, kind, proxyUser)
     new ProcessSession(id, kind, proxyUser, process)
     new ProcessSession(id, kind, proxyUser, process)
   }
   }
 
 
   // Loop until we've started a process with a valid port.
   // Loop until we've started a process with a valid port.
-  private def startProcess(livyConf: LivyConf, id: String, kind: Session.Kind, proxyUser: Option[String]): Process = {
+  private def startProcess(livyConf: LivyConf, id: String, kind: Kind, proxyUser: Option[String]): Process = {
 
 
     val builder = new SparkProcessBuilder()
     val builder = new SparkProcessBuilder()
 
 
@@ -54,7 +55,7 @@ object ProcessSession extends Logging {
 }
 }
 
 
 private class ProcessSession(id: String,
 private class ProcessSession(id: String,
-                             kind: Session.Kind,
+                             kind: Kind,
                              proxyUser: Option[String],
                              proxyUser: Option[String],
                              process: Process) extends WebSession(id, kind, proxyUser) {
                              process: Process) extends WebSession(id, kind, proxyUser) {
 
 

+ 1 - 39
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/Session.scala

@@ -5,56 +5,18 @@ import java.util.concurrent.TimeoutException
 
 
 import com.cloudera.hue.livy.Utils
 import com.cloudera.hue.livy.Utils
 import com.cloudera.hue.livy.msgs.ExecuteRequest
 import com.cloudera.hue.livy.msgs.ExecuteRequest
-import com.cloudera.hue.livy.server.sessions.Statement
+import com.cloudera.hue.livy.sessions.{Kind, State}
 
 
 import scala.concurrent._
 import scala.concurrent._
 import scala.concurrent.duration.Duration
 import scala.concurrent.duration.Duration
 
 
 object Session {
 object Session {
-  sealed trait State
-
-  case class NotStarted() extends State {
-    override def toString = "not_started"
-  }
-
-  case class Starting() extends State {
-    override def toString = "starting"
-  }
-
-  case class Idle() extends State {
-    override def toString = "idle"
-  }
-
-  case class Busy() extends State {
-    override def toString = "busy"
-  }
-
-  case class Error() extends State {
-    override def toString = "error"
-  }
-
-  case class Dead() extends State {
-    override def toString = "dead"
-  }
-
-  sealed trait Kind
-
-  case class Spark() extends Kind {
-    override def toString = "spark"
-  }
-
-  case class PySpark() extends Kind {
-    override def toString = "pyspark"
-  }
-
   class SessionFailedToStart(msg: String) extends Exception(msg)
   class SessionFailedToStart(msg: String) extends Exception(msg)
 
 
   class StatementNotFound extends Exception
   class StatementNotFound extends Exception
 }
 }
 
 
 trait Session {
 trait Session {
-  import Session._
-
   def id: String
   def id: String
 
 
   def kind: Kind
   def kind: Kind

+ 5 - 4
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/SessionFactory.scala

@@ -3,12 +3,13 @@ package com.cloudera.hue.livy.server.sessions
 import java.util.UUID
 import java.util.UUID
 
 
 import com.cloudera.hue.livy.LivyConf
 import com.cloudera.hue.livy.LivyConf
+import com.cloudera.hue.livy.sessions.Kind
 import com.cloudera.hue.livy.yarn.Client
 import com.cloudera.hue.livy.yarn.Client
 
 
 import scala.concurrent.{ExecutionContext, Future}
 import scala.concurrent.{ExecutionContext, Future}
 
 
 trait SessionFactory {
 trait SessionFactory {
-  def createSession(kind: Session.Kind, proxyUser: Option[String] = None): Future[Session]
+  def createSession(kind: Kind, proxyUser: Option[String] = None): Future[Session]
 
 
   def close(): Unit = {}
   def close(): Unit = {}
 }
 }
@@ -17,7 +18,7 @@ class ThreadSessionFactory(livyConf: LivyConf) extends SessionFactory {
 
 
   implicit def executor: ExecutionContext = ExecutionContext.global
   implicit def executor: ExecutionContext = ExecutionContext.global
 
 
-  override def createSession(kind: Session.Kind, proxyUser: Option[String] = None): Future[Session] = {
+  override def createSession(kind: Kind, proxyUser: Option[String] = None): Future[Session] = {
     Future {
     Future {
       val id = UUID.randomUUID().toString
       val id = UUID.randomUUID().toString
       ThreadSession.create(id, kind)
       ThreadSession.create(id, kind)
@@ -29,7 +30,7 @@ class ProcessSessionFactory(livyConf: LivyConf) extends SessionFactory {
 
 
   implicit def executor: ExecutionContext = ExecutionContext.global
   implicit def executor: ExecutionContext = ExecutionContext.global
 
 
-  override def createSession(kind: Session.Kind, proxyUser: Option[String] = None): Future[Session] = {
+  override def createSession(kind: Kind, proxyUser: Option[String] = None): Future[Session] = {
     Future {
     Future {
       val id = UUID.randomUUID().toString
       val id = UUID.randomUUID().toString
       ProcessSession.create(livyConf, id, kind, proxyUser)
       ProcessSession.create(livyConf, id, kind, proxyUser)
@@ -41,7 +42,7 @@ class YarnSessionFactory(livyConf: LivyConf) extends SessionFactory {
 
 
   val client = new Client(livyConf)
   val client = new Client(livyConf)
 
 
-  override def createSession(kind: Session.Kind, proxyUser: Option[String] = None): Future[Session] = {
+  override def createSession(kind: Kind, proxyUser: Option[String] = None): Future[Session] = {
     val id = UUID.randomUUID().toString
     val id = UUID.randomUUID().toString
     YarnSession.create(client, id, kind, proxyUser)
     YarnSession.create(client, id, kind, proxyUser)
   }
   }

+ 2 - 1
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/SessionManager.scala

@@ -1,6 +1,7 @@
 package com.cloudera.hue.livy.server.sessions
 package com.cloudera.hue.livy.server.sessions
 
 
 import com.cloudera.hue.livy.Logging
 import com.cloudera.hue.livy.Logging
+import com.cloudera.hue.livy.sessions.Kind
 
 
 import scala.collection.concurrent.TrieMap
 import scala.collection.concurrent.TrieMap
 import scala.concurrent.duration.Duration
 import scala.concurrent.duration.Duration
@@ -35,7 +36,7 @@ class SessionManager(factory: SessionFactory) extends Logging {
     sessions.keys
     sessions.keys
   }
   }
 
 
-  def createSession(kind: Session.Kind, proxyUser: Option[String] = None): Future[Session] = {
+  def createSession(kind: Kind, proxyUser: Option[String] = None): Future[Session] = {
     val session = factory.createSession(kind, proxyUser = proxyUser)
     val session = factory.createSession(kind, proxyUser = proxyUser)
 
 
     session.map({ case(session: Session) =>
     session.map({ case(session: Session) =>

+ 24 - 23
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/WebApp.scala → apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/SessionServlet.scala

@@ -6,6 +6,7 @@ import java.util.concurrent.TimeUnit
 import com.cloudera.hue.livy.Logging
 import com.cloudera.hue.livy.Logging
 import com.cloudera.hue.livy.msgs.ExecuteRequest
 import com.cloudera.hue.livy.msgs.ExecuteRequest
 import com.cloudera.hue.livy.server.sessions.Session.SessionFailedToStart
 import com.cloudera.hue.livy.server.sessions.Session.SessionFailedToStart
+import com.cloudera.hue.livy.sessions._
 import com.fasterxml.jackson.core.JsonParseException
 import com.fasterxml.jackson.core.JsonParseException
 import org.json4s.JsonAST.JString
 import org.json4s.JsonAST.JString
 import org.json4s._
 import org.json4s._
@@ -15,9 +16,9 @@ import org.scalatra.json.JacksonJsonSupport
 import scala.concurrent._
 import scala.concurrent._
 import scala.concurrent.duration._
 import scala.concurrent.duration._
 
 
-object WebApp extends Logging
+object SessionServlet extends Logging
 
 
-class WebApp(sessionManager: SessionManager)
+class SessionServlet(sessionManager: SessionManager)
   extends ScalatraServlet
   extends ScalatraServlet
   with FutureSupport
   with FutureSupport
   with MethodOverride
   with MethodOverride
@@ -31,20 +32,20 @@ class WebApp(sessionManager: SessionManager)
     contentType = formats("json")
     contentType = formats("json")
   }
   }
 
 
-  get("/sessions") {
+  get("/") {
     Map(
     Map(
       "sessions" -> sessionManager.getSessions
       "sessions" -> sessionManager.getSessions
     )
     )
   }
   }
 
 
-  val getSession = get("/sessions/:sessionId") {
+  val getSession = get("/:sessionId") {
     sessionManager.get(params("sessionId")) match {
     sessionManager.get(params("sessionId")) match {
       case Some(session) => session
       case Some(session) => session
       case None => NotFound("Session not found")
       case None => NotFound("Session not found")
     }
     }
   }
   }
 
 
-  post("/sessions") {
+  post("/") {
     val createSessionRequest = parsedBody.extract[CreateSessionRequest]
     val createSessionRequest = parsedBody.extract[CreateSessionRequest]
     val sessionFuture = sessionManager.createSession(createSessionRequest.lang, createSessionRequest.proxyUser)
     val sessionFuture = sessionManager.createSession(createSessionRequest.lang, createSessionRequest.proxyUser)
 
 
@@ -59,12 +60,12 @@ class WebApp(sessionManager: SessionManager)
     new AsyncResult { val is = rep }
     new AsyncResult { val is = rep }
   }
   }
 
 
-  post("/sessions/:sessionId/callback") {
+  post("/:sessionId/callback") {
     val callback = parsedBody.extract[CallbackRequest]
     val callback = parsedBody.extract[CallbackRequest]
 
 
     sessionManager.get(params("sessionId")) match {
     sessionManager.get(params("sessionId")) match {
       case Some(session) =>
       case Some(session) =>
-        if (session.state == Session.Starting()) {
+        if (session.state == Starting()) {
           session.url = new URL(callback.url)
           session.url = new URL(callback.url)
           Accepted()
           Accepted()
         } else {
         } else {
@@ -74,7 +75,7 @@ class WebApp(sessionManager: SessionManager)
     }
     }
   }
   }
 
 
-  post("/sessions/:sessionId/stop") {
+  post("/:sessionId/stop") {
     sessionManager.get(params("sessionId")) match {
     sessionManager.get(params("sessionId")) match {
       case Some(session) =>
       case Some(session) =>
         val future = session.stop()
         val future = session.stop()
@@ -84,7 +85,7 @@ class WebApp(sessionManager: SessionManager)
     }
     }
   }
   }
 
 
-  post("/sessions/:sessionId/interrupt") {
+  post("/:sessionId/interrupt") {
     sessionManager.get(params("sessionId")) match {
     sessionManager.get(params("sessionId")) match {
       case Some(session) =>
       case Some(session) =>
         val future = for {
         val future = for {
@@ -97,7 +98,7 @@ class WebApp(sessionManager: SessionManager)
     }
     }
   }
   }
 
 
-  delete("/sessions/:sessionId") {
+  delete("/:sessionId") {
     val future = for {
     val future = for {
       _ <- sessionManager.delete(params("sessionId"))
       _ <- sessionManager.delete(params("sessionId"))
     } yield Accepted()
     } yield Accepted()
@@ -105,7 +106,7 @@ class WebApp(sessionManager: SessionManager)
     new AsyncResult() { val is = for { _ <- future } yield NoContent() }
     new AsyncResult() { val is = for { _ <- future } yield NoContent() }
   }
   }
 
 
-  get("/sessions/:sessionId/statements") {
+  get("/:sessionId/statements") {
     sessionManager.get(params("sessionId")) match {
     sessionManager.get(params("sessionId")) match {
       case Some(session: Session) =>
       case Some(session: Session) =>
         Map(
         Map(
@@ -115,7 +116,7 @@ class WebApp(sessionManager: SessionManager)
     }
     }
   }
   }
 
 
-  val getStatement = get("/sessions/:sessionId/statements/:statementId") {
+  val getStatement = get("/:sessionId/statements/:statementId") {
     sessionManager.get(params("sessionId")) match {
     sessionManager.get(params("sessionId")) match {
       case Some(session) =>
       case Some(session) =>
         session.statement(params("statementId").toInt) match {
         session.statement(params("statementId").toInt) match {
@@ -126,7 +127,7 @@ class WebApp(sessionManager: SessionManager)
     }
     }
   }
   }
 
 
-  post("/sessions/:sessionId/statements") {
+  post("/:sessionId/statements") {
     val req = parsedBody.extract[ExecuteRequest]
     val req = parsedBody.extract[ExecuteRequest]
 
 
     sessionManager.get(params("sessionId")) match {
     sessionManager.get(params("sessionId")) match {
@@ -148,12 +149,12 @@ class WebApp(sessionManager: SessionManager)
     case e: SessionFailedToStart => InternalServerError(e.getMessage)
     case e: SessionFailedToStart => InternalServerError(e.getMessage)
     case e: dispatch.StatusCode => ActionResult(ResponseStatus(e.code), e.getMessage, Map.empty)
     case e: dispatch.StatusCode => ActionResult(ResponseStatus(e.code), e.getMessage, Map.empty)
     case e =>
     case e =>
-      WebApp.error("internal error", e)
+      SessionServlet.error("internal error", e)
       InternalServerError(e.toString)
       InternalServerError(e.toString)
   }
   }
 }
 }
 
 
-private case class CreateSessionRequest(lang: Session.Kind, proxyUser: Option[String])
+private case class CreateSessionRequest(lang: Kind, proxyUser: Option[String])
 private case class CallbackRequest(url: String)
 private case class CallbackRequest(url: String)
 
 
 private object Serializers {
 private object Serializers {
@@ -163,9 +164,9 @@ private object Serializers {
   def StatementFormats: List[CustomSerializer[_]] = List(StatementSerializer, StatementStateSerializer)
   def StatementFormats: List[CustomSerializer[_]] = List(StatementSerializer, StatementStateSerializer)
   def Formats: List[CustomSerializer[_]] = SessionFormats ++ StatementFormats
   def Formats: List[CustomSerializer[_]] = SessionFormats ++ StatementFormats
 
 
-  private def serializeSessionState(state: Session.State) = JString(state.toString)
+  private def serializeSessionState(state: State) = JString(state.toString)
 
 
-  private def serializeSessionKind(kind: Session.Kind) = JString(kind.toString)
+  private def serializeSessionKind(kind: Kind) = JString(kind.toString)
 
 
   private def serializeStatementState(state: Statement.State) = JString(state.toString)
   private def serializeStatementState(state: Statement.State) = JString(state.toString)
 
 
@@ -184,20 +185,20 @@ private object Serializers {
     )
     )
   )
   )
 
 
-  case object SessionKindSerializer extends CustomSerializer[Session.Kind](implicit formats => ( {
-    case JString("spark") | JString("scala") => Session.Spark()
-    case JString("pyspark") | JString("python") => Session.PySpark()
+  case object SessionKindSerializer extends CustomSerializer[Kind](implicit formats => ( {
+    case JString("spark") | JString("scala") => Spark()
+    case JString("pyspark") | JString("python") => PySpark()
   }, {
   }, {
-    case kind: Session.Kind => serializeSessionKind(kind)
+    case kind: Kind => serializeSessionKind(kind)
   }
   }
     )
     )
   )
   )
 
 
-  case object SessionStateSerializer extends CustomSerializer[Session.State](implicit formats => ( {
+  case object SessionStateSerializer extends CustomSerializer[State](implicit formats => ( {
     // We don't support deserialization.
     // We don't support deserialization.
     PartialFunction.empty
     PartialFunction.empty
   }, {
   }, {
-    case state: Session.State => JString(state.toString)
+    case state: State => JString(state.toString)
   }
   }
     )
     )
   )
   )

+ 6 - 18
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/ThreadSession.scala

@@ -3,11 +3,9 @@ package com.cloudera.hue.livy.server.sessions
 import java.net.URL
 import java.net.URL
 
 
 import com.cloudera.hue.livy.msgs.ExecuteRequest
 import com.cloudera.hue.livy.msgs.ExecuteRequest
-import com.cloudera.hue.livy.repl
 import com.cloudera.hue.livy.repl.python.PythonSession
 import com.cloudera.hue.livy.repl.python.PythonSession
 import com.cloudera.hue.livy.repl.scala.SparkSession
 import com.cloudera.hue.livy.repl.scala.SparkSession
-import com.cloudera.hue.livy.server.sessions.Session._
-import com.cloudera.hue.livy.server.sessions.Statement
+import com.cloudera.hue.livy.sessions.{Kind, PySpark, Spark, State}
 
 
 import scala.collection.mutable.ArrayBuffer
 import scala.collection.mutable.ArrayBuffer
 import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, Future}
 import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, Future}
@@ -16,11 +14,11 @@ object ThreadSession {
   val LIVY_HOME = System.getenv("LIVY_HOME")
   val LIVY_HOME = System.getenv("LIVY_HOME")
   val LIVY_REPL = LIVY_HOME + "/bin/livy-repl"
   val LIVY_REPL = LIVY_HOME + "/bin/livy-repl"
 
 
-  def create(id: String, kind: Session.Kind): Session = {
+  def create(id: String, kind: Kind): Session = {
     val session = kind match {
     val session = kind match {
-      case Session.Spark() =>
+      case Spark() =>
         SparkSession.create()
         SparkSession.create()
-      case Session.PySpark() =>
+      case PySpark() =>
         PythonSession.createPySpark()
         PythonSession.createPySpark()
     }
     }
     new ThreadSession(id, kind, session)
     new ThreadSession(id, kind, session)
@@ -28,7 +26,7 @@ object ThreadSession {
 }
 }
 
 
 private class ThreadSession(val id: String,
 private class ThreadSession(val id: String,
-                            val kind: Session.Kind,
+                            val kind: Kind,
                             session: com.cloudera.hue.livy.repl.Session) extends Session {
                             session: com.cloudera.hue.livy.repl.Session) extends Session {
 
 
   protected implicit def executor: ExecutionContextExecutor = ExecutionContext.global
   protected implicit def executor: ExecutionContextExecutor = ExecutionContext.global
@@ -40,17 +38,7 @@ private class ThreadSession(val id: String,
 
 
   override def lastActivity: Long = 0
   override def lastActivity: Long = 0
 
 
-  override def state: State = {
-    session.state match {
-      case repl.Session.NotStarted() => NotStarted()
-      case repl.Session.Starting() => Starting()
-      case repl.Session.Idle() => Idle()
-      case repl.Session.Busy() => Busy()
-      case repl.Session.ShuttingDown() => Dead()
-      case repl.Session.ShutDown() => Dead()
-      case repl.Session.Error() => Error()
-    }
-  }
+  override def state: State = session.state
 
 
   override def url: Option[URL] = None
   override def url: Option[URL] = None
 
 

+ 7 - 3
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/WebSession.scala

@@ -5,8 +5,7 @@ import java.util.concurrent.TimeUnit
 
 
 import com.cloudera.hue.livy._
 import com.cloudera.hue.livy._
 import com.cloudera.hue.livy.msgs.ExecuteRequest
 import com.cloudera.hue.livy.msgs.ExecuteRequest
-import com.cloudera.hue.livy.server.sessions.Session._
-import com.cloudera.hue.livy.server.sessions.Statement
+import com.cloudera.hue.livy.sessions._
 import dispatch._
 import dispatch._
 import org.json4s.jackson.Serialization.write
 import org.json4s.jackson.Serialization.write
 import org.json4s.{DefaultFormats, Formats}
 import org.json4s.{DefaultFormats, Formats}
@@ -33,7 +32,7 @@ class WebSession(val id: String,
   override def url: Option[URL] = _url
   override def url: Option[URL] = _url
 
 
   override def url_=(url: URL) = {
   override def url_=(url: URL) = {
-    ensureState(Session.Starting(), {
+    ensureState(Starting(), {
       _state = Idle()
       _state = Idle()
       _url = Some(url)
       _url = Some(url)
     })
     })
@@ -112,6 +111,11 @@ class WebSession(val id: String,
             waitForStateChange(Busy(), Duration(10, TimeUnit.SECONDS))
             waitForStateChange(Busy(), Duration(10, TimeUnit.SECONDS))
             stop()
             stop()
           }
           }
+        case ShuttingDown() =>
+          Future {
+            waitForStateChange(ShuttingDown(), Duration(10, TimeUnit.SECONDS))
+            stop()
+          }
         case Error() | Dead() =>
         case Error() | Dead() =>
           Future.successful(Unit)
           Future.successful(Unit)
       }
       }

+ 5 - 4
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/sessions/YarnSession.scala

@@ -2,6 +2,7 @@ package com.cloudera.hue.livy.server.sessions
 
 
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.TimeUnit
 
 
+import com.cloudera.hue.livy.sessions.{Kind, Error}
 import com.cloudera.hue.livy.yarn.{Client, Job}
 import com.cloudera.hue.livy.yarn.{Client, Job}
 
 
 import scala.concurrent.{Await, ExecutionContext, ExecutionContextExecutor, Future}
 import scala.concurrent.{Await, ExecutionContext, ExecutionContextExecutor, Future}
@@ -10,7 +11,7 @@ import scala.concurrent.duration._
 object YarnSession {
 object YarnSession {
   protected implicit def executor: ExecutionContextExecutor = ExecutionContext.global
   protected implicit def executor: ExecutionContextExecutor = ExecutionContext.global
 
 
-  def create(client: Client, id: String, kind: Session.Kind, proxyUser: Option[String] = None): Future[Session] = {
+  def create(client: Client, id: String, kind: Kind, proxyUser: Option[String] = None): Future[Session] = {
     val callbackUrl = System.getProperty("livy.server.callback-url")
     val callbackUrl = System.getProperty("livy.server.callback-url")
     val job = client.submitApplication(
     val job = client.submitApplication(
       id = id,
       id = id,
@@ -23,11 +24,11 @@ object YarnSession {
 }
 }
 
 
 private class YarnSession(id: String,
 private class YarnSession(id: String,
-                          kind: Session.Kind,
+                          kind: Kind,
                           proxyUser: Option[String],
                           proxyUser: Option[String],
                           job: Future[Job]) extends WebSession(id, kind, proxyUser) {
                           job: Future[Job]) extends WebSession(id, kind, proxyUser) {
   job.onFailure { case _ =>
   job.onFailure { case _ =>
-    _state = Session.Error()
+    _state = Error()
   }
   }
 
 
   override def stop(): Future[Unit] = {
   override def stop(): Future[Unit] = {
@@ -38,7 +39,7 @@ private class YarnSession(id: String,
           job_.waitForFinish(10000)
           job_.waitForFinish(10000)
         } catch {
         } catch {
           case e: Throwable =>
           case e: Throwable =>
-            _state = Session.Error()
+            _state = Error()
             throw e
             throw e
         }
         }
     }
     }

+ 6 - 5
apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/BaseSessionSpec.scala

@@ -4,6 +4,7 @@ import java.util.concurrent.TimeUnit
 
 
 import com.cloudera.hue.livy.msgs.ExecuteRequest
 import com.cloudera.hue.livy.msgs.ExecuteRequest
 import com.cloudera.hue.livy.server.sessions.Session
 import com.cloudera.hue.livy.server.sessions.Session
+import com.cloudera.hue.livy.sessions.{Starting, Idle}
 import org.json4s.{DefaultFormats, Extraction}
 import org.json4s.{DefaultFormats, Extraction}
 import org.scalatest.{BeforeAndAfter, FunSpec, Matchers}
 import org.scalatest.{BeforeAndAfter, FunSpec, Matchers}
 
 
@@ -28,16 +29,16 @@ abstract class BaseSessionSpec extends FunSpec with Matchers with BeforeAndAfter
 
 
   describe("A spark session") {
   describe("A spark session") {
     it("should start in the starting or idle state") {
     it("should start in the starting or idle state") {
-      session.state should (equal (Session.Starting()) or equal (Session.Idle()))
+      session.state should (equal (Starting()) or equal (Idle()))
     }
     }
 
 
     it("should eventually become the idle state") {
     it("should eventually become the idle state") {
-      session.waitForStateChange(Session.Starting(), Duration(30, TimeUnit.SECONDS))
-      session.state should equal (Session.Idle())
+      session.waitForStateChange(Starting(), Duration(30, TimeUnit.SECONDS))
+      session.state should equal (Idle())
     }
     }
 
 
     it("should execute `1 + 2` == 3") {
     it("should execute `1 + 2` == 3") {
-      session.waitForStateChange(Session.Starting(), Duration(30, TimeUnit.SECONDS))
+      session.waitForStateChange(Starting(), Duration(30, TimeUnit.SECONDS))
       val stmt = session.executeStatement(ExecuteRequest("1 + 2"))
       val stmt = session.executeStatement(ExecuteRequest("1 + 2"))
       val result = Await.result(stmt.output, Duration.Inf)
       val result = Await.result(stmt.output, Duration.Inf)
 
 
@@ -53,7 +54,7 @@ abstract class BaseSessionSpec extends FunSpec with Matchers with BeforeAndAfter
     }
     }
 
 
     it("should report an error if accessing an unknown variable") {
     it("should report an error if accessing an unknown variable") {
-      session.waitForStateChange(Session.Starting(), Duration(30, TimeUnit.SECONDS))
+      session.waitForStateChange(Starting(), Duration(30, TimeUnit.SECONDS))
       val stmt = session.executeStatement(ExecuteRequest("x"))
       val stmt = session.executeStatement(ExecuteRequest("x"))
       val result = Await.result(stmt.output, Duration.Inf)
       val result = Await.result(stmt.output, Duration.Inf)
       val expectedResult = Extraction.decompose(Map(
       val expectedResult = Extraction.decompose(Map(

+ 4 - 4
apps/spark/java/livy-server/src/test/scala/com/cloudera/hue/livy/server/ProcessSessionSpec.scala

@@ -1,14 +1,14 @@
 package com.cloudera.hue.livy.server
 package com.cloudera.hue.livy.server
 
 
 import com.cloudera.hue.livy.LivyConf
 import com.cloudera.hue.livy.LivyConf
-import com.cloudera.hue.livy.server.sessions.{ProcessSession, Session}
-import org.scalatest.matchers.ShouldMatchers
-import org.scalatest.{Matchers, FunSpecLike, BeforeAndAfter, FunSpec}
+import com.cloudera.hue.livy.server.sessions.ProcessSession
+import com.cloudera.hue.livy.sessions.Spark
+import org.scalatest.{BeforeAndAfter, FunSpecLike, Matchers}
 
 
 class ProcessSessionSpec extends BaseSessionSpec with FunSpecLike with Matchers with BeforeAndAfter {
 class ProcessSessionSpec extends BaseSessionSpec with FunSpecLike with Matchers with BeforeAndAfter {
 
 
   val livyConf = new LivyConf()
   val livyConf = new LivyConf()
   livyConf.set("livy.repl.driverClassPath", sys.props("java.class.path"))
   livyConf.set("livy.repl.driverClassPath", sys.props("java.class.path"))
 
 
-  def createSession() = ProcessSession.create(livyConf, "0", Session.Spark())
+  def createSession() = ProcessSession.create(livyConf, "0", Spark())
 }
 }

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

@@ -1,9 +1,10 @@
 package com.cloudera.hue.livy.server
 package com.cloudera.hue.livy.server
 
 
-import com.cloudera.hue.livy.server.sessions.{Session, ThreadSession}
+import com.cloudera.hue.livy.server.sessions.ThreadSession
+import com.cloudera.hue.livy.sessions.Spark
 import org.scalatest.{BeforeAndAfter, FunSpecLike, Matchers}
 import org.scalatest.{BeforeAndAfter, FunSpecLike, Matchers}
 
 
 class ThreadSessionSpec extends BaseSessionSpec with FunSpecLike with Matchers with BeforeAndAfter {
 class ThreadSessionSpec extends BaseSessionSpec with FunSpecLike with Matchers with BeforeAndAfter {
 
 
-  def createSession() = ThreadSession.create("0", Session.Spark())
+  def createSession() = ThreadSession.create("0", Spark())
 }
 }