From 8487bbbb155a132366f03b2d928cc2a0f9b94813 Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam(mfathisalmi)" <mfathisalmi@paypal.com>
Date: Tue, 12 Sep 2017 11:18:25 -0700
Subject: [PATCH 01/24] [LIVY-41] Let users access sessions by session name

This commit  enables Livy users to access sessions either by names or by auto-generated sessiond id's.

It also prevents users from creating sessions that have the same name.

This commit keeps API change  minimal. Thse are palces that API change
is needed:
- `Session` and its subclasses adds a new field, `name`.
- `RecoveryMetadata` and its subclasses adds a new field, `name`.
- `SessionManager` adds a new method `getSession(name: String)` which lookups sessions by name.

A more clean implementation would change the signature of `SessionManager.register` so it returns a proper container around the session value to determine if it failed to register the given session.  For example,
```Scala
def register(session S): Either[S, Throwable]
 ```

Task-url: https://issues.apache.org/jira/browse/LIVY-41
---
 .../livy/client/common/HttpMessages.java      |  8 ++--
 .../livy/test/framework/LivyRestClient.scala  |  4 ++
 .../scala/org/apache/livy/test/BatchIT.scala  |  4 +-
 .../org/apache/livy/test/InteractiveIT.scala  |  2 +-
 .../apache/livy/server/SessionServlet.scala   | 14 +++++--
 .../livy/server/batch/BatchSession.scala      |  9 ++++-
 .../server/batch/BatchSessionServlet.scala    | 21 ++++++++++-
 .../interactive/InteractiveSession.scala      | 11 ++++--
 .../InteractiveSessionServlet.scala           | 22 +++++++++--
 .../org/apache/livy/sessions/Session.scala    |  2 +-
 .../apache/livy/sessions/SessionManager.scala | 21 ++++++++++-
 .../livy/server/SessionServletSpec.scala      |  2 +-
 .../livy/server/batch/BatchServletSpec.scala  |  4 ++
 .../livy/server/batch/BatchSessionSpec.scala  |  7 ++--
 .../InteractiveSessionServletSpec.scala       |  3 ++
 .../interactive/InteractiveSessionSpec.scala  |  7 ++--
 .../interactive/SessionHeartbeatSpec.scala    |  5 ++-
 .../apache/livy/sessions/MockSession.scala    |  3 +-
 .../livy/sessions/SessionManagerSpec.scala    | 37 ++++++++++++++++++-
 19 files changed, 155 insertions(+), 31 deletions(-)

diff --git a/client-common/src/main/java/org/apache/livy/client/common/HttpMessages.java b/client-common/src/main/java/org/apache/livy/client/common/HttpMessages.java
index b1e253fb0..2245eb982 100644
--- a/client-common/src/main/java/org/apache/livy/client/common/HttpMessages.java
+++ b/client-common/src/main/java/org/apache/livy/client/common/HttpMessages.java
@@ -53,6 +53,7 @@ private CreateClientRequest() {
   public static class SessionInfo implements ClientMessage {
 
     public final int id;
+    public final String name;
     public final String appId;
     public final String owner;
     public final String proxyUser;
@@ -61,9 +62,10 @@ public static class SessionInfo implements ClientMessage {
     public final Map<String, String> appInfo;
     public final List<String> log;
 
-    public SessionInfo(int id, String appId, String owner, String proxyUser, String state,
-        String kind, Map<String, String> appInfo, List<String> log) {
+    public SessionInfo(int id, String name, String appId, String owner, String proxyUser,
+        String state, String kind, Map<String, String> appInfo, List<String> log) {
       this.id = id;
+      this.name = name;
       this.appId = appId;
       this.owner = owner;
       this.proxyUser = proxyUser;
@@ -74,7 +76,7 @@ public SessionInfo(int id, String appId, String owner, String proxyUser, String
     }
 
     private SessionInfo() {
-      this(-1, null, null, null, null, null, null, null);
+      this(-1, null, null, null, null, null, null, null, null);
     }
 
   }
diff --git a/integration-test/src/main/scala/org/apache/livy/test/framework/LivyRestClient.scala b/integration-test/src/main/scala/org/apache/livy/test/framework/LivyRestClient.scala
index 6d319c757..206e79521 100644
--- a/integration-test/src/main/scala/org/apache/livy/test/framework/LivyRestClient.scala
+++ b/integration-test/src/main/scala/org/apache/livy/test/framework/LivyRestClient.scala
@@ -205,12 +205,14 @@ class LivyRestClient(val httpClient: AsyncHttpClient, val livyEndpoint: String)
   }
 
   def startBatch(
+      name: String,
       file: String,
       className: Option[String],
       args: List[String],
       sparkConf: Map[String, String]): BatchSession = {
     val r = new CreateBatchRequest()
     r.file = file
+    r.name = Option(name)
     r.className = className
     r.args = args
     r.conf = Map("spark.yarn.maxAppAttempts" -> "1") ++ sparkConf
@@ -220,12 +222,14 @@ class LivyRestClient(val httpClient: AsyncHttpClient, val livyEndpoint: String)
   }
 
   def startSession(
+      name: String,
       kind: Kind,
       sparkConf: Map[String, String],
       heartbeatTimeoutInSecond: Int): InteractiveSession = {
     val r = new CreateInteractiveRequest()
     r.kind = kind
     r.conf = sparkConf
+    r.name = Option(name)
     r.heartbeatTimeoutInSecond = heartbeatTimeoutInSecond
 
     val id = start(INTERACTIVE_TYPE, mapper.writeValueAsString(r))
diff --git a/integration-test/src/test/scala/org/apache/livy/test/BatchIT.scala b/integration-test/src/test/scala/org/apache/livy/test/BatchIT.scala
index 0b3a06194..acc96f8ae 100644
--- a/integration-test/src/test/scala/org/apache/livy/test/BatchIT.scala
+++ b/integration-test/src/test/scala/org/apache/livy/test/BatchIT.scala
@@ -161,14 +161,14 @@ class BatchIT extends BaseIntegrationTestSuite with BeforeAndAfterAll {
   private def withScript[R]
     (scriptPath: String, args: List[String], sparkConf: Map[String, String] = Map.empty)
     (f: (LivyRestClient#BatchSession) => R): R = {
-    val s = livyClient.startBatch(scriptPath, None, args, sparkConf)
+    val s = livyClient.startBatch(null, scriptPath, None, args, sparkConf)
     withSession(s)(f)
   }
 
   private def withTestLib[R]
     (testClass: Class[_], args: List[String], sparkConf: Map[String, String] = Map.empty)
     (f: (LivyRestClient#BatchSession) => R): R = {
-    val s = livyClient.startBatch(testLibPath, Some(testClass.getName()), args, sparkConf)
+    val s = livyClient.startBatch(null, testLibPath, Some(testClass.getName()), args, sparkConf)
     withSession(s)(f)
   }
 }
diff --git a/integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala b/integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala
index 728fd1c2c..1225f3843 100644
--- a/integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala
+++ b/integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala
@@ -191,7 +191,7 @@ class InteractiveIT extends BaseIntegrationTestSuite {
       waitForIdle: Boolean = true,
       heartbeatTimeoutInSecond: Int = 0)
     (f: (LivyRestClient#InteractiveSession) => R): R = {
-    withSession(livyClient.startSession(kind, sparkConf, heartbeatTimeoutInSecond)) { s =>
+    withSession(livyClient.startSession(null, kind, sparkConf, heartbeatTimeoutInSecond)) { s =>
       if (waitForIdle) {
         s.verifySessionIdle()
       }
diff --git a/server/src/main/scala/org/apache/livy/server/SessionServlet.scala b/server/src/main/scala/org/apache/livy/server/SessionServlet.scala
index 76b5afab0..ffffd32f1 100644
--- a/server/src/main/scala/org/apache/livy/server/SessionServlet.scala
+++ b/server/src/main/scala/org/apache/livy/server/SessionServlet.scala
@@ -22,6 +22,7 @@ import javax.servlet.http.HttpServletRequest
 import org.scalatra._
 import scala.concurrent._
 import scala.concurrent.duration._
+import scala.util.{Failure, Success, Try}
 
 import org.apache.livy.{LivyConf, Logging}
 import org.apache.livy.sessions.{Session, SessionManager}
@@ -214,8 +215,15 @@ abstract class SessionServlet[S <: Session, R <: RecoveryMetadata](
   private def doWithSession(fn: (S => Any),
       allowAll: Boolean,
       checkFn: Option[(String, HttpServletRequest) => Boolean]): Any = {
-    val sessionId = params("id").toInt
-    sessionManager.get(sessionId) match {
+    val idParam: String = params("id")
+    val session = if (idParam.forall(_.isDigit)) {
+      val sessionId = idParam.toInt
+      sessionManager.get(sessionId)
+    } else {
+      val sessionName = idParam
+      sessionManager.get(sessionName)
+    }
+    session match {
       case Some(session) =>
         if (allowAll || checkFn.map(_(session.owner, request)).getOrElse(false)) {
           fn(session)
@@ -223,7 +231,7 @@ abstract class SessionServlet[S <: Session, R <: RecoveryMetadata](
           Forbidden()
         }
       case None =>
-        NotFound(s"Session '$sessionId' not found.")
+        NotFound(s"Session '$idParam' not found.")
     }
   }
 
diff --git a/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala b/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
index 2605bf50c..3a17d669a 100644
--- a/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
@@ -33,6 +33,7 @@ import org.apache.livy.utils.{AppInfo, SparkApp, SparkAppListener, SparkProcessB
 @JsonIgnoreProperties(ignoreUnknown = true)
 case class BatchRecoveryMetadata(
     id: Int,
+    name: String,
     appId: Option[String],
     appTag: String,
     owner: String,
@@ -45,6 +46,7 @@ object BatchSession extends Logging {
 
   def create(
       id: Int,
+      name: String,
       request: CreateBatchRequest,
       livyConf: LivyConf,
       owner: String,
@@ -92,6 +94,7 @@ object BatchSession extends Logging {
 
     new BatchSession(
       id,
+      name,
       appTag,
       SessionState.Starting(),
       livyConf,
@@ -108,6 +111,7 @@ object BatchSession extends Logging {
       mockApp: Option[SparkApp] = None): BatchSession = {
     new BatchSession(
       m.id,
+      m.name,
       m.appTag,
       SessionState.Recovering(),
       livyConf,
@@ -122,6 +126,7 @@ object BatchSession extends Logging {
 
 class BatchSession(
     id: Int,
+    name: String,
     appTag: String,
     initialState: SessionState,
     livyConf: LivyConf,
@@ -129,7 +134,7 @@ class BatchSession(
     override val proxyUser: Option[String],
     sessionStore: SessionStore,
     sparkApp: BatchSession => SparkApp)
-  extends Session(id, owner, livyConf) with SparkAppListener {
+  extends Session(id, name, owner, livyConf) with SparkAppListener {
   import BatchSession._
 
   protected implicit def executor: ExecutionContextExecutor = ExecutionContext.global
@@ -169,5 +174,5 @@ class BatchSession(
   override def infoChanged(appInfo: AppInfo): Unit = { this.appInfo = appInfo }
 
   override def recoveryMetadata: RecoveryMetadata =
-    BatchRecoveryMetadata(id, appId, appTag, owner, proxyUser)
+    BatchRecoveryMetadata(id, name, appId, appTag, owner, proxyUser)
 }
diff --git a/server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala b/server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala
index 85945d988..8a0b49164 100644
--- a/server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala
+++ b/server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala
@@ -27,6 +27,7 @@ import org.apache.livy.utils.AppInfo
 
 case class BatchSessionView(
   id: Long,
+  name: String,
   state: String,
   appId: Option[String],
   appInfo: AppInfo,
@@ -43,8 +44,23 @@ class BatchSessionServlet(
   override protected def createSession(req: HttpServletRequest): BatchSession = {
     val createRequest = bodyAs[CreateBatchRequest](req)
     val proxyUser = checkImpersonation(createRequest.proxyUser, req)
+    val sessionId = sessionManager.nextId()
+    val sessionName: String = createRequest.name match {
+      case Some(name) if sessionManager.get(name).isEmpty =>
+        name
+      case Some(name) =>
+        // this does NOT guarantee that by the time this session is ready to be registered in
+        // sessionManager, another with the same name is not registered. But in most cases,
+        // it prevents Livy from submitting applications to Spark.
+        val msg = s"Session $name already exists! " +
+          s"Choose a different name or delete the existing session."
+        throw new IllegalArgumentException(msg)
+      case None =>
+        s"INTERACTIVE-SESSION-$sessionId"
+    }
+
     BatchSession.create(
-      sessionManager.nextId(), createRequest, livyConf, remoteUser(req), proxyUser, sessionStore)
+      sessionId, sessionName, createRequest, livyConf, remoteUser(req), proxyUser, sessionStore)
   }
 
   override protected[batch] def clientSessionView(
@@ -62,7 +78,8 @@ class BatchSessionServlet(
       } else {
         Nil
       }
-    BatchSessionView(session.id, session.state.toString, session.appId, session.appInfo, logs)
+    BatchSessionView(session.id, session.name, session.state.toString, session.appId,
+      session.appInfo, logs)
   }
 
 }
diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
index 0462e80fc..05fe7fae8 100644
--- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
@@ -48,6 +48,7 @@ import org.apache.livy.utils._
 @JsonIgnoreProperties(ignoreUnknown = true)
 case class InteractiveRecoveryMetadata(
     id: Int,
+    name: String,
     appId: Option[String],
     appTag: String,
     kind: Kind,
@@ -65,6 +66,7 @@ object InteractiveSession extends Logging {
 
   def create(
       id: Int,
+      name: String,
       owner: String,
       proxyUser: Option[String],
       livyConf: LivyConf,
@@ -109,6 +111,7 @@ object InteractiveSession extends Logging {
 
     new InteractiveSession(
       id,
+      name,
       None,
       appTag,
       client,
@@ -135,6 +138,7 @@ object InteractiveSession extends Logging {
 
     new InteractiveSession(
       metadata.id,
+      metadata.name,
       metadata.appId,
       metadata.appTag,
       client,
@@ -345,6 +349,7 @@ object InteractiveSession extends Logging {
 
 class InteractiveSession(
     id: Int,
+    name: String,
     appIdHint: Option[String],
     appTag: String,
     client: Option[RSCClient],
@@ -356,7 +361,7 @@ class InteractiveSession(
     override val proxyUser: Option[String],
     sessionStore: SessionStore,
     mockApp: Option[SparkApp]) // For unit test.
-  extends Session(id, owner, livyConf)
+  extends Session(id, name, owner, livyConf)
   with SessionHeartbeat
   with SparkAppListener {
 
@@ -436,8 +441,8 @@ class InteractiveSession(
   override def logLines(): IndexedSeq[String] = app.map(_.log()).getOrElse(sessionLog)
 
   override def recoveryMetadata: RecoveryMetadata =
-    InteractiveRecoveryMetadata(
-      id, appId, appTag, kind, heartbeatTimeout.toSeconds.toInt, owner, proxyUser, rscDriverUri)
+    InteractiveRecoveryMetadata( id, name, appId, appTag, kind, heartbeatTimeout.toSeconds.toInt,
+      owner, proxyUser, rscDriverUri)
 
   override def state: SessionState = {
     if (serverSideState.isInstanceOf[SessionState.Running]) {
diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala
index 38008568c..e128cc616 100644
--- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala
+++ b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala
@@ -53,8 +53,23 @@ class InteractiveSessionServlet(
   override protected def createSession(req: HttpServletRequest): InteractiveSession = {
     val createRequest = bodyAs[CreateInteractiveRequest](req)
     val proxyUser = checkImpersonation(createRequest.proxyUser, req)
+    val sessionId: Int = sessionManager.nextId()
+    val sessionName: String = createRequest.name match {
+      case Some(name) if sessionManager.get(name).isEmpty =>
+        name
+      case Some(name) =>
+        // this does NOT guarantee that by the time this session is ready to be registered in
+        // sessionManager, another with the same name is not registered. But in most cases,
+        // it prevents Livy from submitting applications to Spark.
+        val msg = s"Session $name already exists! " +
+          s"Choose a different name or delete the existing session."
+        throw new IllegalArgumentException(msg)
+      case None =>
+        s"INTERACTIVE-SESSION-$sessionId"
+    }
     InteractiveSession.create(
-      sessionManager.nextId(),
+      sessionId,
+      sessionName,
       remoteUser(req),
       proxyUser,
       livyConf,
@@ -80,8 +95,9 @@ class InteractiveSessionServlet(
         Nil
       }
 
-    new SessionInfo(session.id, session.appId.orNull, session.owner, session.proxyUser.orNull,
-      session.state.toString, session.kind.toString, session.appInfo.asJavaMap, logs.asJava)
+    new SessionInfo(session.id, session.name, session.appId.orNull, session.owner,
+      session.proxyUser.orNull, session.state.toString, session.kind.toString,
+      session.appInfo.asJavaMap, logs.asJava)
   }
 
   post("/:id/stop") {
diff --git a/server/src/main/scala/org/apache/livy/sessions/Session.scala b/server/src/main/scala/org/apache/livy/sessions/Session.scala
index d467076fc..87b884c1f 100644
--- a/server/src/main/scala/org/apache/livy/sessions/Session.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/Session.scala
@@ -132,7 +132,7 @@ object Session {
   }
 }
 
-abstract class Session(val id: Int, val owner: String, val livyConf: LivyConf)
+abstract class Session(val id: Int, val name: String, val owner: String, val livyConf: LivyConf)
   extends Logging {
 
   import Session._
diff --git a/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala b/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
index d482c335d..e1591db0a 100644
--- a/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
@@ -73,6 +73,8 @@ class SessionManager[S <: Session, R <: RecoveryMetadata : ClassTag](
 
   protected[this] final val idCounter = new AtomicInteger(0)
   protected[this] final val sessions = mutable.LinkedHashMap[Int, S]()
+  private[this] final val sessionsByName = mutable.Map[String, S]()
+
 
   private[this] final val sessionTimeoutCheck = livyConf.getBoolean(LivyConf.SESSION_TIMEOUT_CHECK)
   private[this] final val sessionTimeout =
@@ -92,13 +94,29 @@ class SessionManager[S <: Session, R <: RecoveryMetadata : ClassTag](
   def register(session: S): S = {
     info(s"Registering new session ${session.id}")
     synchronized {
-      sessions.put(session.id, session)
+      // in InteractiveSessionServlet.createSession() and BatchSessionServlet.createSession(),
+      // Livy checks to make sure another session with the same name does not exist already.
+      // This case should not happen rarely.
+      // already exists. But looking up a session name and adding it to the set of existing sessions
+      // should happen atomically.
+      if (sessionsByName.contains(session.name)) {
+        val msg = s"Session ${session.name} already exists!"
+        error(msg)
+        delete(session)
+        throw new IllegalArgumentException(msg)
+      } else {
+        info(s"sessionNames = ${sessionsByName.keys.mkString}")
+        sessions.put(session.id, session)
+        sessionsByName.put(session.name, session)
+      }
     }
     session
   }
 
   def get(id: Int): Option[S] = sessions.get(id)
 
+  def get(sessionName: String): Option[S] = sessionsByName.get(sessionName)
+
   def size(): Int = sessions.size
 
   def all(): Iterable[S] = sessions.values
@@ -113,6 +131,7 @@ class SessionManager[S <: Session, R <: RecoveryMetadata : ClassTag](
         sessionStore.remove(sessionType, session.id)
         synchronized {
           sessions.remove(session.id)
+          sessionsByName.remove(session.name)
         }
       } catch {
         case NonFatal(e) =>
diff --git a/server/src/test/scala/org/apache/livy/server/SessionServletSpec.scala b/server/src/test/scala/org/apache/livy/server/SessionServletSpec.scala
index 2410fe275..ff82d0b2a 100644
--- a/server/src/test/scala/org/apache/livy/server/SessionServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/SessionServletSpec.scala
@@ -32,7 +32,7 @@ object SessionServletSpec {
   val PROXY_USER = "proxyUser"
 
   class MockSession(id: Int, owner: String, livyConf: LivyConf)
-    extends Session(id, owner, livyConf) {
+    extends Session(id, s"Mock Session $id", owner, livyConf) {
 
     case class MockRecoveryMetadata(id: Int) extends RecoveryMetadata()
 
diff --git a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
index 866044832..f77f397d0 100644
--- a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
@@ -116,6 +116,7 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
 
     it("should respect config black list") {
       val createRequest = new CreateBatchRequest()
+      createRequest.name = Some("TEST-BatchServletSpec-Session-0")
       createRequest.file = script.toString
       createRequest.conf = BLACKLISTED_CONFIG
       jpost[Map[String, Any]]("/", createRequest, expectedStatus = SC_BAD_REQUEST) { _ => }
@@ -123,6 +124,7 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
 
     it("should show session properties") {
       val id = 0
+      val name = "TEST-batch-session"
       val state = SessionState.Running()
       val appId = "appid"
       val appInfo = AppInfo(Some("DRIVER LOG URL"), Some("SPARK UI URL"))
@@ -130,6 +132,7 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
 
       val session = mock[BatchSession]
       when(session.id).thenReturn(id)
+      when(session.name).thenReturn(name)
       when(session.state).thenReturn(state)
       when(session.appId).thenReturn(Some(appId))
       when(session.appInfo).thenReturn(appInfo)
@@ -141,6 +144,7 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
         .asInstanceOf[BatchSessionView]
 
       view.id shouldEqual id
+      view.name shouldEqual name
       view.state shouldEqual state.toString
       view.appId shouldEqual Some(appId)
       view.appInfo shouldEqual appInfo
diff --git a/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala b/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala
index eb80bef90..84bdcf346 100644
--- a/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala
@@ -68,7 +68,7 @@ class BatchSessionSpec
       req.conf = Map("spark.driver.extraClassPath" -> sys.props("java.class.path"))
 
       val conf = new LivyConf().set(LivyConf.LOCAL_FS_WHITELIST, sys.props("java.io.tmpdir"))
-      val batch = BatchSession.create(0, req, conf, null, None, sessionStore)
+      val batch = BatchSession.create(0, "Test Batch Session", req, conf, null, None, sessionStore)
 
       Utils.waitUntil({ () => !batch.state.isActive }, Duration(10, TimeUnit.SECONDS))
       (batch.state match {
@@ -83,7 +83,8 @@ class BatchSessionSpec
       val conf = new LivyConf()
       val req = new CreateBatchRequest()
       val mockApp = mock[SparkApp]
-      val batch = BatchSession.create(0, req, conf, null, None, sessionStore, Some(mockApp))
+      val batch = BatchSession.create(
+        0, "Test Batch Session", req, conf, null, None, sessionStore, Some(mockApp))
 
       val expectedAppId = "APPID"
       batch.appIdKnown(expectedAppId)
@@ -100,7 +101,7 @@ class BatchSessionSpec
       val conf = new LivyConf()
       val req = new CreateBatchRequest()
       val mockApp = mock[SparkApp]
-      val m = BatchRecoveryMetadata(99, None, "appTag", null, None)
+      val m = BatchRecoveryMetadata(99, "Test Batch Session", None, "appTag", null, None)
       val batch = BatchSession.recover(m, conf, sessionStore, Some(mockApp))
 
       batch.state shouldBe a[SessionState.Recovering]
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
index e1de22e3b..802517998 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
@@ -149,6 +149,7 @@ class InteractiveSessionServletSpec extends BaseInteractiveServletSpec {
 
   it("should show session properties") {
     val id = 0
+    val name = "TEST-interactive-session"
     val appId = "appid"
     val owner = "owner"
     val proxyUser = "proxyUser"
@@ -159,6 +160,7 @@ class InteractiveSessionServletSpec extends BaseInteractiveServletSpec {
 
     val session = mock[InteractiveSession]
     when(session.id).thenReturn(id)
+    when(session.name).thenReturn(name)
     when(session.appId).thenReturn(Some(appId))
     when(session.owner).thenReturn(owner)
     when(session.proxyUser).thenReturn(Some(proxyUser))
@@ -173,6 +175,7 @@ class InteractiveSessionServletSpec extends BaseInteractiveServletSpec {
       .asInstanceOf[SessionInfo]
 
     view.id shouldEqual id
+    view.name shouldEqual name
     view.appId shouldEqual appId
     view.owner shouldEqual owner
     view.proxyUser shouldEqual proxyUser
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala
index 9943c00cd..9e5f400f7 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala
@@ -68,7 +68,8 @@ class InteractiveSessionSpec extends FunSpec
       SparkLauncher.DRIVER_EXTRA_CLASSPATH -> sys.props("java.class.path"),
       RSCConf.Entry.LIVY_JARS.key() -> ""
     )
-    InteractiveSession.create(0, null, None, livyConf, req, sessionStore, mockApp)
+    InteractiveSession.create(
+      0, "Test Interactive Session", null, None, livyConf, req, sessionStore, mockApp)
   }
 
   private def executeStatement(code: String, codeType: Option[String] = None): JValue = {
@@ -247,7 +248,7 @@ class InteractiveSessionSpec extends FunSpec
       when(mockClient.submit(any(classOf[PingJob]))).thenReturn(mock[JobHandle[Void]])
       val m =
         InteractiveRecoveryMetadata(
-          78, None, "appTag", Spark(), 0, null, None, Some(URI.create("")))
+          78, "Test session ", None, "appTag", Spark(), 0, null, None, Some(URI.create("")))
       val s = InteractiveSession.recover(m, conf, sessionStore, None, Some(mockClient))
 
       s.state shouldBe a[SessionState.Recovering]
@@ -261,7 +262,7 @@ class InteractiveSessionSpec extends FunSpec
       val conf = new LivyConf()
       val sessionStore = mock[SessionStore]
       val m = InteractiveRecoveryMetadata(
-        78, Some("appId"), "appTag", Spark(), 0, null, None, None)
+        78, "Test session ", Some("appId"), "appTag", Spark(), 0, null, None, None)
       val s = InteractiveSession.recover(m, conf, sessionStore, None)
 
       s.state shouldBe a[SessionState.Dead]
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/SessionHeartbeatSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/SessionHeartbeatSpec.scala
index 12c8bbb24..86576b79e 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/SessionHeartbeatSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/SessionHeartbeatSpec.scala
@@ -55,7 +55,8 @@ class SessionHeartbeatSpec extends FunSpec with Matchers {
   }
 
   describe("SessionHeartbeatWatchdog") {
-    abstract class TestSession extends Session(0, null, null) with SessionHeartbeat {}
+    abstract class TestSession
+      extends Session(0, "Test Heart Beat Session", null, null) with SessionHeartbeat {}
     class TestWatchdog(conf: LivyConf)
       extends SessionManager[TestSession, RecoveryMetadata](
         conf,
@@ -68,10 +69,12 @@ class SessionHeartbeatSpec extends FunSpec with Matchers {
     it("should delete only expired sessions") {
       val expiredSession: TestSession = mock[TestSession]
       when(expiredSession.id).thenReturn(0)
+      when(expiredSession.name).thenReturn("expired session")
       when(expiredSession.heartbeatExpired).thenReturn(true)
 
       val nonExpiredSession: TestSession = mock[TestSession]
       when(nonExpiredSession.id).thenReturn(1)
+      when(nonExpiredSession.name).thenReturn("unexpired session")
       when(nonExpiredSession.heartbeatExpired).thenReturn(false)
 
       val n = new TestWatchdog(new LivyConf())
diff --git a/server/src/test/scala/org/apache/livy/sessions/MockSession.scala b/server/src/test/scala/org/apache/livy/sessions/MockSession.scala
index 3cfbe466b..5e3c4665c 100644
--- a/server/src/test/scala/org/apache/livy/sessions/MockSession.scala
+++ b/server/src/test/scala/org/apache/livy/sessions/MockSession.scala
@@ -19,7 +19,8 @@ package org.apache.livy.sessions
 
 import org.apache.livy.LivyConf
 
-class MockSession(id: Int, owner: String, conf: LivyConf) extends Session(id, owner, conf) {
+class MockSession(id: Int, owner: String, conf: LivyConf, name: String = s"Mock-Session")
+  extends Session(id, name, owner, conf) {
   case class RecoveryMetadata(id: Int) extends Session.RecoveryMetadata()
 
   override val proxyUser = None
diff --git a/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala b/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
index beffa7136..354d873ac 100644
--- a/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
+++ b/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
@@ -54,6 +54,41 @@ class SessionManagerSpec extends FunSpec with Matchers with LivyBaseUnitTestSuit
       }
     }
 
+    it("should create sessions with names") {
+      val livyConf = new LivyConf()
+      livyConf.set(LivyConf.SESSION_TIMEOUT, "100ms")
+      val manager = new SessionManager[MockSession, RecoveryMetadata](
+        livyConf,
+        { _ => assert(false).asInstanceOf[MockSession] },
+        mock[SessionStore],
+        "test",
+        Some(Seq.empty))
+      val session = manager.register(new MockSession(manager.nextId(), null, livyConf, "session1"))
+      manager.get(session.id).isDefined should be(true)
+      manager.get(session.name).isDefined should be(true)
+    }
+
+    it("should not create sessions with the same name") {
+      val livyConf = new LivyConf()
+      livyConf.set(LivyConf.SESSION_TIMEOUT, "100ms")
+      val manager = new SessionManager[MockSession, RecoveryMetadata](
+        livyConf,
+        { _ => assert(false).asInstanceOf[MockSession] },
+        mock[SessionStore],
+        "test",
+        Some(Seq.empty))
+      val session1 = new MockSession(manager.nextId(), null, livyConf, "test session name")
+      val session2 = new MockSession(manager.nextId(), null, livyConf, "test session name")
+      manager.register(session1)
+      an[IllegalArgumentException] should be thrownBy manager.register(session2)
+      manager.get(session1.id).isDefined should be(true)
+      manager.get(session2.id).isDefined should be(false)
+      eventually(timeout(5 seconds), interval(100 millis)) {
+        Await.result(manager.collectGarbage(), Duration.Inf)
+        manager.get(session1.id) should be(None)
+      }
+    }
+
     it("batch session should not be gc-ed until application is finished") {
       val sessionId = 24
       val session = mock[BatchSession]
@@ -112,7 +147,7 @@ class SessionManagerSpec extends FunSpec with Matchers with LivyBaseUnitTestSuit
     implicit def executor: ExecutionContext = ExecutionContext.global
 
     def makeMetadata(id: Int, appTag: String): BatchRecoveryMetadata = {
-      BatchRecoveryMetadata(id, None, appTag, null, None)
+      BatchRecoveryMetadata(id, s"test-session-$id", None, appTag, null, None)
     }
 
     def mockSession(id: Int): BatchSession = {

From 3f28b611ec7b587dbca2ff69cc99a7654d44b430 Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam(mfathisalmi)" <mfathisalmi@paypal.com>
Date: Thu, 5 Oct 2017 16:15:01 -0700
Subject: [PATCH 02/24] [LIVY-41] renaming id tp idOrName

Signed-off-by: Fathi Salmi, Meisam(mfathisalmi) <mfathisalmi@paypal.com>
---
 .../scala/org/apache/livy/server/SessionServlet.scala | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/server/src/main/scala/org/apache/livy/server/SessionServlet.scala b/server/src/main/scala/org/apache/livy/server/SessionServlet.scala
index ffffd32f1..460979047 100644
--- a/server/src/main/scala/org/apache/livy/server/SessionServlet.scala
+++ b/server/src/main/scala/org/apache/livy/server/SessionServlet.scala
@@ -120,6 +120,7 @@ abstract class SessionServlet[S <: Session, R <: RecoveryMetadata](
 
   post("/") {
     val session = sessionManager.register(createSession(request))
+    session.startSession
     // Because it may take some time to establish the session, update the last activity
     // time before returning the session info to the client.
     session.recordActivity()
@@ -215,12 +216,12 @@ abstract class SessionServlet[S <: Session, R <: RecoveryMetadata](
   private def doWithSession(fn: (S => Any),
       allowAll: Boolean,
       checkFn: Option[(String, HttpServletRequest) => Boolean]): Any = {
-    val idParam: String = params("id")
-    val session = if (idParam.forall(_.isDigit)) {
-      val sessionId = idParam.toInt
+    val idOrNameParam: String = params("id")
+    val session = if (idOrNameParam.forall(_.isDigit)) {
+      val sessionId = idOrNameParam.toInt
       sessionManager.get(sessionId)
     } else {
-      val sessionName = idParam
+      val sessionName = idOrNameParam
       sessionManager.get(sessionName)
     }
     session match {
@@ -231,7 +232,7 @@ abstract class SessionServlet[S <: Session, R <: RecoveryMetadata](
           Forbidden()
         }
       case None =>
-        NotFound(s"Session '$idParam' not found.")
+        NotFound(s"Session '$idOrNameParam' not found.")
     }
   }
 

From dc644c7fbdb6da6ba2d573df9ea29cc2e04ff785 Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam(mfathisalmi)" <mfathisalmi@paypal.com>
Date: Thu, 5 Oct 2017 16:23:52 -0700
Subject: [PATCH 03/24] [LIVY-41] refactoring guareded case statements

Changing
```
case ... if .... =>
```
to
```
case ... =>
  if ...
```
---
 .../server/batch/BatchSessionServlet.scala    |  20 +--
 .../interactive/InteractiveSession.scala      | 119 +++++++++---------
 2 files changed, 73 insertions(+), 66 deletions(-)

diff --git a/server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala b/server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala
index 8a0b49164..5db477f67 100644
--- a/server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala
+++ b/server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala
@@ -46,17 +46,19 @@ class BatchSessionServlet(
     val proxyUser = checkImpersonation(createRequest.proxyUser, req)
     val sessionId = sessionManager.nextId()
     val sessionName: String = createRequest.name match {
-      case Some(name) if sessionManager.get(name).isEmpty =>
-        name
       case Some(name) =>
-        // this does NOT guarantee that by the time this session is ready to be registered in
-        // sessionManager, another with the same name is not registered. But in most cases,
-        // it prevents Livy from submitting applications to Spark.
-        val msg = s"Session $name already exists! " +
-          s"Choose a different name or delete the existing session."
-        throw new IllegalArgumentException(msg)
+        if (sessionManager.get(name).isEmpty)
+          name
+        else {
+          // this does NOT guarantee that by the time this session is ready to be registered in
+          // sessionManager, another with the same name is not registered. But in most cases,
+          // it prevents Livy from submitting applications to Spark.
+          val msg = s"Session $name already exists! " +
+            s"Choose a different name or delete the existing session."
+          throw new IllegalArgumentException(msg)
+        }
       case None =>
-        s"INTERACTIVE-SESSION-$sessionId"
+        s"batch-$sessionId"
     }
 
     BatchSession.create(
diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
index 05fe7fae8..9a4356c5f 100644
--- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
@@ -26,15 +26,14 @@ import java.util.concurrent.atomic.AtomicLong
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable
-import scala.concurrent.Future
+import scala.concurrent.{Future, Promise}
 import scala.concurrent.duration.{Duration, FiniteDuration}
-import scala.util.Random
+import scala.util.{Random, Try}
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties
 import com.google.common.annotations.VisibleForTesting
 import org.apache.hadoop.fs.Path
 import org.apache.spark.launcher.SparkLauncher
-
 import org.apache.livy._
 import org.apache.livy.client.common.HttpMessages._
 import org.apache.livy.rsc.{PingJob, RSCClient, RSCConf}
@@ -383,62 +382,10 @@ class InteractiveSession(
   sessionStore.save(RECOVERY_SESSION_TYPE, recoveryMetadata)
   heartbeat()
 
-  private val app = mockApp.orElse {
-    val driverProcess = client.flatMap { c => Option(c.getDriverProcess) }
-        .map(new LineBufferedProcess(_, livyConf.getInt(LivyConf.SPARK_LOGS_SIZE)))
-    driverProcess.map { _ => SparkApp.create(appTag, appId, driverProcess, livyConf, Some(this)) }
-  }
-
-  if (client.isEmpty) {
-    transition(Dead())
-    val msg = s"Cannot recover interactive session $id because its RSCDriver URI is unknown."
-    info(msg)
-    sessionLog = IndexedSeq(msg)
-  } else {
-    val uriFuture = Future { client.get.getServerUri.get() }
-
-    uriFuture onSuccess { case url =>
-      rscDriverUri = Option(url)
-      sessionSaveLock.synchronized {
-        sessionStore.save(RECOVERY_SESSION_TYPE, recoveryMetadata)
-      }
-    }
-    uriFuture onFailure { case e => warn("Fail to get rsc uri", e) }
-
-    // Send a dummy job that will return once the client is ready to be used, and set the
-    // state to "idle" at that point.
-    client.get.submit(new PingJob()).addListener(new JobHandle.Listener[Void]() {
-      override def onJobQueued(job: JobHandle[Void]): Unit = { }
-      override def onJobStarted(job: JobHandle[Void]): Unit = { }
-
-      override def onJobCancelled(job: JobHandle[Void]): Unit = errorOut()
-
-      override def onJobFailed(job: JobHandle[Void], cause: Throwable): Unit = errorOut()
+  private var app :Option[SparkApp] = None
 
-      override def onJobSucceeded(job: JobHandle[Void], result: Void): Unit = {
-        transition(SessionState.Running())
-        info(s"Interactive session $id created [appid: ${appId.orNull}, owner: $owner, proxyUser:" +
-          s" $proxyUser, state: ${state.toString}, kind: ${kind.toString}, " +
-          s"info: ${appInfo.asJavaMap}]")
-      }
 
-      private def errorOut(): Unit = {
-        // Other code might call stop() to close the RPC channel. When RPC channel is closing,
-        // this callback might be triggered. Check and don't call stop() to avoid nested called
-        // if the session is already shutting down.
-        if (serverSideState != SessionState.ShuttingDown()) {
-          transition(SessionState.Error())
-          stop()
-          app.foreach { a =>
-            info(s"Failed to ping RSC driver for session $id. Killing application.")
-            a.kill()
-          }
-        }
-      }
-    })
-  }
-
-  override def logLines(): IndexedSeq[String] = app.map(_.log()).getOrElse(sessionLog)
+  override def logLines(): IndexedSeq[String] = app.trySuccess()map(_.log()).getOrElse(sessionLog)
 
   override def recoveryMetadata: RecoveryMetadata =
     InteractiveRecoveryMetadata( id, name, appId, appTag, kind, heartbeatTimeout.toSeconds.toInt,
@@ -456,6 +403,64 @@ class InteractiveSession(
     }
   }
 
+  override def startSession(): Unit = {
+    app = mockApp.orElse {
+      val driverProcess = client.flatMap { c => Option(c.getDriverProcess) }
+        .map(new LineBufferedProcess(_, livyConf.getInt(LivyConf.SPARK_LOGS_SIZE)))
+      driverProcess.map { _ => SparkApp.create(appTag, appId, driverProcess, livyConf, Some(this))
+      }
+    }
+
+    if (client.isEmpty) {
+      transition(Dead())
+      val msg = s"Cannot recover interactive session $id because its RSCDriver URI is unknown."
+      info(msg)
+      sessionLog = IndexedSeq(msg)
+    } else {
+      val uriFuture = Future { client.get.getServerUri.get() }
+
+      uriFuture onSuccess { case url =>
+        rscDriverUri = Option(url)
+        sessionSaveLock.synchronized {
+          sessionStore.save(RECOVERY_SESSION_TYPE, recoveryMetadata)
+        }
+      }
+      uriFuture onFailure { case e => warn("Fail to get rsc uri", e) }
+
+      // Send a dummy job that will return once the client is ready to be used, and set the
+      // state to "idle" at that point.
+      client.get.submit(new PingJob()).addListener(new JobHandle.Listener[Void]() {
+        override def onJobQueued(job: JobHandle[Void]): Unit = { }
+        override def onJobStarted(job: JobHandle[Void]): Unit = { }
+
+        override def onJobCancelled(job: JobHandle[Void]): Unit = errorOut()
+
+        override def onJobFailed(job: JobHandle[Void], cause: Throwable): Unit = errorOut()
+
+        override def onJobSucceeded(job: JobHandle[Void], result: Void): Unit = {
+          transition(SessionState.Running())
+          info(s"Interactive session $id created [appid: ${appId.orNull}, owner: $owner, proxyUser:" +
+            s" $proxyUser, state: ${state.toString}, kind: ${kind.toString}, " +
+            s"info: ${appInfo.asJavaMap}]")
+        }
+
+        private def errorOut(): Unit = {
+          // Other code might call stop() to close the RPC channel. When RPC channel is closing,
+          // this callback might be triggered. Check and don't call stop() to avoid nested called
+          // if the session is already shutting down.
+          if (serverSideState != SessionState.ShuttingDown()) {
+            transition(SessionState.Error())
+            stop()
+            app.foreach { a =>
+              info(s"Failed to ping RSC driver for session $id. Killing application.")
+              a.kill()
+            }
+          }
+        }
+      })
+    }
+  }
+
   override def stopSession(): Unit = {
     try {
       transition(SessionState.ShuttingDown())

From 967e896a0c5c4344de57dace5bc4fa62c9f3698a Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam(mfathisalmi)" <mfathisalmi@paypal.com>
Date: Mon, 9 Oct 2017 20:30:17 -0700
Subject: [PATCH 04/24] First shot at making brnach names optional

---
 .../livy/test/framework/LivyRestClient.scala  |  8 ++--
 .../apache/livy/server/SessionServlet.scala   |  1 -
 .../livy/server/batch/BatchSession.scala      | 20 ++++++----
 .../server/batch/BatchSessionServlet.scala    | 19 +--------
 .../interactive/InteractiveSession.scala      | 20 +++++-----
 .../InteractiveSessionServlet.scala           | 16 +-------
 .../org/apache/livy/sessions/Session.scala    | 10 +++--
 .../apache/livy/sessions/SessionManager.scala | 28 ++++++-------
 .../livy/server/SessionServletSpec.scala      |  4 +-
 .../livy/server/batch/BatchServletSpec.scala  |  2 +-
 .../livy/server/batch/BatchSessionSpec.scala  | 25 +++++++++---
 .../InteractiveSessionServletSpec.scala       | 40 ++++++++++++++++++-
 .../interactive/InteractiveSessionSpec.scala  | 33 +++++++++++----
 .../interactive/SessionHeartbeatSpec.scala    |  6 +--
 .../apache/livy/sessions/MockSession.scala    |  4 +-
 .../livy/sessions/SessionManagerSpec.scala    | 12 +++---
 16 files changed, 152 insertions(+), 96 deletions(-)

diff --git a/integration-test/src/main/scala/org/apache/livy/test/framework/LivyRestClient.scala b/integration-test/src/main/scala/org/apache/livy/test/framework/LivyRestClient.scala
index 206e79521..ed4b0d0b5 100644
--- a/integration-test/src/main/scala/org/apache/livy/test/framework/LivyRestClient.scala
+++ b/integration-test/src/main/scala/org/apache/livy/test/framework/LivyRestClient.scala
@@ -205,14 +205,14 @@ class LivyRestClient(val httpClient: AsyncHttpClient, val livyEndpoint: String)
   }
 
   def startBatch(
-      name: String,
+      name: Option[String],
       file: String,
       className: Option[String],
       args: List[String],
       sparkConf: Map[String, String]): BatchSession = {
     val r = new CreateBatchRequest()
     r.file = file
-    r.name = Option(name)
+    r.name = name
     r.className = className
     r.args = args
     r.conf = Map("spark.yarn.maxAppAttempts" -> "1") ++ sparkConf
@@ -222,14 +222,14 @@ class LivyRestClient(val httpClient: AsyncHttpClient, val livyEndpoint: String)
   }
 
   def startSession(
-      name: String,
+      name: Option[String],
       kind: Kind,
       sparkConf: Map[String, String],
       heartbeatTimeoutInSecond: Int): InteractiveSession = {
     val r = new CreateInteractiveRequest()
     r.kind = kind
     r.conf = sparkConf
-    r.name = Option(name)
+    r.name = name
     r.heartbeatTimeoutInSecond = heartbeatTimeoutInSecond
 
     val id = start(INTERACTIVE_TYPE, mapper.writeValueAsString(r))
diff --git a/server/src/main/scala/org/apache/livy/server/SessionServlet.scala b/server/src/main/scala/org/apache/livy/server/SessionServlet.scala
index 460979047..c300aa3dc 100644
--- a/server/src/main/scala/org/apache/livy/server/SessionServlet.scala
+++ b/server/src/main/scala/org/apache/livy/server/SessionServlet.scala
@@ -120,7 +120,6 @@ abstract class SessionServlet[S <: Session, R <: RecoveryMetadata](
 
   post("/") {
     val session = sessionManager.register(createSession(request))
-    session.startSession
     // Because it may take some time to establish the session, update the last activity
     // time before returning the session info to the client.
     session.recordActivity()
diff --git a/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala b/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
index 3a17d669a..e1bd6e679 100644
--- a/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
@@ -33,7 +33,7 @@ import org.apache.livy.utils.{AppInfo, SparkApp, SparkAppListener, SparkProcessB
 @JsonIgnoreProperties(ignoreUnknown = true)
 case class BatchRecoveryMetadata(
     id: Int,
-    name: String,
+    name: Option[String],
     appId: Option[String],
     appTag: String,
     owner: String,
@@ -46,7 +46,7 @@ object BatchSession extends Logging {
 
   def create(
       id: Int,
-      name: String,
+      name: Option[String],
       request: CreateBatchRequest,
       livyConf: LivyConf,
       owner: String,
@@ -126,7 +126,7 @@ object BatchSession extends Logging {
 
 class BatchSession(
     id: Int,
-    name: String,
+    name: Option[String],
     appTag: String,
     initialState: SessionState,
     livyConf: LivyConf,
@@ -140,18 +140,24 @@ class BatchSession(
   protected implicit def executor: ExecutionContextExecutor = ExecutionContext.global
 
   private[this] var _state: SessionState = initialState
-  private val app = sparkApp(this)
+
+  private var app: Option[SparkApp] = None
 
   override def state: SessionState = _state
 
-  override def logLines(): IndexedSeq[String] = app.log()
+  override def logLines(): IndexedSeq[String] = app.map(_.log()).getOrElse(IndexedSeq.empty[String])
+
+  override def start(): Unit = {
+    app = Option(sparkApp(this))
+  }
 
   override def stopSession(): Unit = {
-    app.kill()
+    app.foreach(_.kill())
   }
 
   override def appIdKnown(appId: String): Unit = {
-    _appId = Option(appId)
+    _appId = Option(
+      appId)
     sessionStore.save(RECOVERY_SESSION_TYPE, recoveryMetadata)
   }
 
diff --git a/server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala b/server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala
index 5db477f67..a9b8a031f 100644
--- a/server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala
+++ b/server/src/main/scala/org/apache/livy/server/batch/BatchSessionServlet.scala
@@ -27,7 +27,7 @@ import org.apache.livy.utils.AppInfo
 
 case class BatchSessionView(
   id: Long,
-  name: String,
+  name: Option[String],
   state: String,
   appId: Option[String],
   appInfo: AppInfo,
@@ -45,22 +45,7 @@ class BatchSessionServlet(
     val createRequest = bodyAs[CreateBatchRequest](req)
     val proxyUser = checkImpersonation(createRequest.proxyUser, req)
     val sessionId = sessionManager.nextId()
-    val sessionName: String = createRequest.name match {
-      case Some(name) =>
-        if (sessionManager.get(name).isEmpty)
-          name
-        else {
-          // this does NOT guarantee that by the time this session is ready to be registered in
-          // sessionManager, another with the same name is not registered. But in most cases,
-          // it prevents Livy from submitting applications to Spark.
-          val msg = s"Session $name already exists! " +
-            s"Choose a different name or delete the existing session."
-          throw new IllegalArgumentException(msg)
-        }
-      case None =>
-        s"batch-$sessionId"
-    }
-
+    val sessionName = createRequest.name
     BatchSession.create(
       sessionId, sessionName, createRequest, livyConf, remoteUser(req), proxyUser, sessionStore)
   }
diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
index 9a4356c5f..d98c3753e 100644
--- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
@@ -34,6 +34,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties
 import com.google.common.annotations.VisibleForTesting
 import org.apache.hadoop.fs.Path
 import org.apache.spark.launcher.SparkLauncher
+
 import org.apache.livy._
 import org.apache.livy.client.common.HttpMessages._
 import org.apache.livy.rsc.{PingJob, RSCClient, RSCConf}
@@ -47,7 +48,7 @@ import org.apache.livy.utils._
 @JsonIgnoreProperties(ignoreUnknown = true)
 case class InteractiveRecoveryMetadata(
     id: Int,
-    name: String,
+    name: Option[String],
     appId: Option[String],
     appTag: String,
     kind: Kind,
@@ -65,7 +66,7 @@ object InteractiveSession extends Logging {
 
   def create(
       id: Int,
-      name: String,
+      name: Option[String],
       owner: String,
       proxyUser: Option[String],
       livyConf: LivyConf,
@@ -348,7 +349,7 @@ object InteractiveSession extends Logging {
 
 class InteractiveSession(
     id: Int,
-    name: String,
+    name: Option[String],
     appIdHint: Option[String],
     appTag: String,
     client: Option[RSCClient],
@@ -382,10 +383,9 @@ class InteractiveSession(
   sessionStore.save(RECOVERY_SESSION_TYPE, recoveryMetadata)
   heartbeat()
 
-  private var app :Option[SparkApp] = None
-
+  private var app: Option[SparkApp] = None
 
-  override def logLines(): IndexedSeq[String] = app.trySuccess()map(_.log()).getOrElse(sessionLog)
+  override def logLines(): IndexedSeq[String] = app.map(_.log()).getOrElse(sessionLog)
 
   override def recoveryMetadata: RecoveryMetadata =
     InteractiveRecoveryMetadata( id, name, appId, appTag, kind, heartbeatTimeout.toSeconds.toInt,
@@ -403,7 +403,7 @@ class InteractiveSession(
     }
   }
 
-  override def startSession(): Unit = {
+  override def start(): Unit = {
     app = mockApp.orElse {
       val driverProcess = client.flatMap { c => Option(c.getDriverProcess) }
         .map(new LineBufferedProcess(_, livyConf.getInt(LivyConf.SPARK_LOGS_SIZE)))
@@ -439,9 +439,9 @@ class InteractiveSession(
 
         override def onJobSucceeded(job: JobHandle[Void], result: Void): Unit = {
           transition(SessionState.Running())
-          info(s"Interactive session $id created [appid: ${appId.orNull}, owner: $owner, proxyUser:" +
-            s" $proxyUser, state: ${state.toString}, kind: ${kind.toString}, " +
-            s"info: ${appInfo.asJavaMap}]")
+          info(s"Interactive session $id created [appid: ${appId.orNull}, " +
+            s"owner: $owner, proxyUser: $proxyUser, state: ${state.toString}, " +
+            s"kind: ${kind.toString}, info: ${appInfo.asJavaMap}]")
         }
 
         private def errorOut(): Unit = {
diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala
index e128cc616..761a13408 100644
--- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala
+++ b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala
@@ -54,19 +54,7 @@ class InteractiveSessionServlet(
     val createRequest = bodyAs[CreateInteractiveRequest](req)
     val proxyUser = checkImpersonation(createRequest.proxyUser, req)
     val sessionId: Int = sessionManager.nextId()
-    val sessionName: String = createRequest.name match {
-      case Some(name) if sessionManager.get(name).isEmpty =>
-        name
-      case Some(name) =>
-        // this does NOT guarantee that by the time this session is ready to be registered in
-        // sessionManager, another with the same name is not registered. But in most cases,
-        // it prevents Livy from submitting applications to Spark.
-        val msg = s"Session $name already exists! " +
-          s"Choose a different name or delete the existing session."
-        throw new IllegalArgumentException(msg)
-      case None =>
-        s"INTERACTIVE-SESSION-$sessionId"
-    }
+    val sessionName = createRequest.name
     InteractiveSession.create(
       sessionId,
       sessionName,
@@ -95,7 +83,7 @@ class InteractiveSessionServlet(
         Nil
       }
 
-    new SessionInfo(session.id, session.name, session.appId.orNull, session.owner,
+    new SessionInfo(session.id, session.name.orNull, session.appId.orNull, session.owner,
       session.proxyUser.orNull, session.state.toString, session.kind.toString,
       session.appInfo.asJavaMap, logs.asJava)
   }
diff --git a/server/src/main/scala/org/apache/livy/sessions/Session.scala b/server/src/main/scala/org/apache/livy/sessions/Session.scala
index 87b884c1f..563e2d494 100644
--- a/server/src/main/scala/org/apache/livy/sessions/Session.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/Session.scala
@@ -21,7 +21,6 @@ import java.io.InputStream
 import java.net.{URI, URISyntaxException}
 import java.security.PrivilegedExceptionAction
 import java.util.UUID
-import java.util.concurrent.TimeUnit
 
 import scala.concurrent.{ExecutionContext, Future}
 
@@ -29,7 +28,9 @@ import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.hadoop.fs.permission.FsPermission
 import org.apache.hadoop.security.UserGroupInformation
 
-import org.apache.livy.{LivyConf, Logging, Utils}
+import org.apache.livy.LivyConf
+import org.apache.livy.Logging
+import org.apache.livy.Utils
 import org.apache.livy.utils.AppInfo
 
 object Session {
@@ -132,7 +133,8 @@ object Session {
   }
 }
 
-abstract class Session(val id: Int, val name: String, val owner: String, val livyConf: LivyConf)
+abstract class Session(val id: Int, val name: Option[String], val owner: String,
+                       val livyConf: LivyConf)
   extends Logging {
 
   import Session._
@@ -168,6 +170,8 @@ abstract class Session(val id: Int, val name: String, val owner: String, val liv
 
   def state: SessionState
 
+  def start(): Unit
+
   def stop(): Future[Unit] = Future {
     try {
       info(s"Stopping $this...")
diff --git a/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala b/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
index e1591db0a..82f96d894 100644
--- a/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
@@ -73,7 +73,7 @@ class SessionManager[S <: Session, R <: RecoveryMetadata : ClassTag](
 
   protected[this] final val idCounter = new AtomicInteger(0)
   protected[this] final val sessions = mutable.LinkedHashMap[Int, S]()
-  private[this] final val sessionsByName = mutable.Map[String, S]()
+  private[this] final val sessionsByName = mutable.HashMap[String, S]()
 
 
   private[this] final val sessionTimeoutCheck = livyConf.getBoolean(LivyConf.SESSION_TIMEOUT_CHECK)
@@ -94,20 +94,18 @@ class SessionManager[S <: Session, R <: RecoveryMetadata : ClassTag](
   def register(session: S): S = {
     info(s"Registering new session ${session.id}")
     synchronized {
-      // in InteractiveSessionServlet.createSession() and BatchSessionServlet.createSession(),
-      // Livy checks to make sure another session with the same name does not exist already.
-      // This case should not happen rarely.
-      // already exists. But looking up a session name and adding it to the set of existing sessions
-      // should happen atomically.
-      if (sessionsByName.contains(session.name)) {
-        val msg = s"Session ${session.name} already exists!"
-        error(msg)
-        delete(session)
-        throw new IllegalArgumentException(msg)
-      } else {
-        info(s"sessionNames = ${sessionsByName.keys.mkString}")
+      session.name.map { sessionName =>
+        if (sessionsByName.contains(sessionName)) {
+          val msg = s"Session ${session.name} already exists!"
+          error(msg)
+          delete(session)
+          throw new IllegalArgumentException(msg)
+        } else {
+          // info(s"sessionNames = ${sessionsByName.keys.mkString}")
+          sessionsByName.put(sessionName, session)
+        }
         sessions.put(session.id, session)
-        sessionsByName.put(session.name, session)
+        session.start()
       }
     }
     session
@@ -131,7 +129,7 @@ class SessionManager[S <: Session, R <: RecoveryMetadata : ClassTag](
         sessionStore.remove(sessionType, session.id)
         synchronized {
           sessions.remove(session.id)
-          sessionsByName.remove(session.name)
+          session.name.foreach(sessionsByName.remove)
         }
       } catch {
         case NonFatal(e) =>
diff --git a/server/src/test/scala/org/apache/livy/server/SessionServletSpec.scala b/server/src/test/scala/org/apache/livy/server/SessionServletSpec.scala
index ff82d0b2a..8b85829ad 100644
--- a/server/src/test/scala/org/apache/livy/server/SessionServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/SessionServletSpec.scala
@@ -32,7 +32,7 @@ object SessionServletSpec {
   val PROXY_USER = "proxyUser"
 
   class MockSession(id: Int, owner: String, livyConf: LivyConf)
-    extends Session(id, s"Mock Session $id", owner, livyConf) {
+    extends Session(id, None, owner, livyConf) {
 
     case class MockRecoveryMetadata(id: Int) extends RecoveryMetadata()
 
@@ -42,6 +42,8 @@ object SessionServletSpec {
 
     override def state: SessionState = SessionState.Idle()
 
+    override def start(): Unit = ()
+
     override protected def stopSession(): Unit = ()
 
     override def logLines(): IndexedSeq[String] = IndexedSeq("log")
diff --git a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
index f77f397d0..42ff6431e 100644
--- a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
@@ -132,7 +132,7 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
 
       val session = mock[BatchSession]
       when(session.id).thenReturn(id)
-      when(session.name).thenReturn(name)
+      when(session.name).thenReturn(Some(name))
       when(session.state).thenReturn(state)
       when(session.appId).thenReturn(Some(appId))
       when(session.appInfo).thenReturn(appInfo)
diff --git a/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala b/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala
index 84bdcf346..82d5a5841 100644
--- a/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala
@@ -68,7 +68,8 @@ class BatchSessionSpec
       req.conf = Map("spark.driver.extraClassPath" -> sys.props("java.class.path"))
 
       val conf = new LivyConf().set(LivyConf.LOCAL_FS_WHITELIST, sys.props("java.io.tmpdir"))
-      val batch = BatchSession.create(0, "Test Batch Session", req, conf, null, None, sessionStore)
+      val batch = BatchSession.create(0, None, req, conf, null, None, sessionStore)
+      batch.start()
 
       Utils.waitUntil({ () => !batch.state.isActive }, Duration(10, TimeUnit.SECONDS))
       (batch.state match {
@@ -83,9 +84,9 @@ class BatchSessionSpec
       val conf = new LivyConf()
       val req = new CreateBatchRequest()
       val mockApp = mock[SparkApp]
-      val batch = BatchSession.create(
-        0, "Test Batch Session", req, conf, null, None, sessionStore, Some(mockApp))
+      val batch = BatchSession.create( 0, None, req, conf, null, None, sessionStore, Some(mockApp))
 
+      batch.start()
       val expectedAppId = "APPID"
       batch.appIdKnown(expectedAppId)
       verify(sessionStore, atLeastOnce()).save(
@@ -97,11 +98,25 @@ class BatchSessionSpec
       batch.appInfo shouldEqual expectedAppInfo
     }
 
-    it("should recover session") {
+    it("should recover named session") {
       val conf = new LivyConf()
       val req = new CreateBatchRequest()
       val mockApp = mock[SparkApp]
-      val m = BatchRecoveryMetadata(99, "Test Batch Session", None, "appTag", null, None)
+      val m = BatchRecoveryMetadata(99, Some("Test Batch Session"), None, "appTag", null, None)
+      val batch = BatchSession.recover(m, conf, sessionStore, Some(mockApp))
+
+      batch.state shouldBe a[SessionState.Recovering]
+
+      batch.appIdKnown("appId")
+      verify(sessionStore, atLeastOnce()).save(
+        Matchers.eq(BatchSession.RECOVERY_SESSION_TYPE), anyObject())
+    }
+
+    it("should recover session with no name ") {
+      val conf = new LivyConf()
+      val req = new CreateBatchRequest()
+      val mockApp = mock[SparkApp]
+      val m = BatchRecoveryMetadata(999, None, None, "appTag", null, None)
       val batch = BatchSession.recover(m, conf, sessionStore, Some(mockApp))
 
       batch.state shouldBe a[SessionState.Recovering]
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
index 802517998..0fd5e8729 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
@@ -160,7 +160,7 @@ class InteractiveSessionServletSpec extends BaseInteractiveServletSpec {
 
     val session = mock[InteractiveSession]
     when(session.id).thenReturn(id)
-    when(session.name).thenReturn(name)
+    when(session.name).thenReturn(Some(name))
     when(session.appId).thenReturn(Some(appId))
     when(session.owner).thenReturn(owner)
     when(session.proxyUser).thenReturn(Some(proxyUser))
@@ -186,4 +186,42 @@ class InteractiveSessionServletSpec extends BaseInteractiveServletSpec {
     view.log shouldEqual log.asJava
   }
 
+  it("should show session properties for sessions without a name") {
+    val id = 0
+    val appId = "appid"
+    val owner = "owner"
+    val proxyUser = "proxyUser"
+    val state = SessionState.Running()
+    val kind = Spark()
+    val appInfo = AppInfo(Some("DRIVER LOG URL"), Some("SPARK UI URL"))
+    val log = IndexedSeq[String]("log1", "log2")
+
+    val session = mock[InteractiveSession]
+    when(session.id).thenReturn(id)
+    when(session.name).thenReturn(None)
+    when(session.appId).thenReturn(Some(appId))
+    when(session.owner).thenReturn(owner)
+    when(session.proxyUser).thenReturn(Some(proxyUser))
+    when(session.state).thenReturn(state)
+    when(session.kind).thenReturn(kind)
+    when(session.appInfo).thenReturn(appInfo)
+    when(session.logLines()).thenReturn(log)
+
+    val req = mock[HttpServletRequest]
+
+    val view = servlet.asInstanceOf[InteractiveSessionServlet].clientSessionView(session, req)
+      .asInstanceOf[SessionInfo]
+
+    view.id shouldEqual id
+    view.name shouldEqual null
+    view.appId shouldEqual appId
+    view.owner shouldEqual owner
+    view.proxyUser shouldEqual proxyUser
+    view.state shouldEqual state.toString
+    view.kind shouldEqual kind.toString
+    view.appInfo should contain (Entry(AppInfo.DRIVER_LOG_URL_NAME, appInfo.driverLogUrl.get))
+    view.appInfo should contain (Entry(AppInfo.SPARK_UI_URL_NAME, appInfo.sparkUiUrl.get))
+    view.log shouldEqual log.asJava
+  }
+
 }
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala
index 9e5f400f7..7bf33f964 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala
@@ -68,8 +68,7 @@ class InteractiveSessionSpec extends FunSpec
       SparkLauncher.DRIVER_EXTRA_CLASSPATH -> sys.props("java.class.path"),
       RSCConf.Entry.LIVY_JARS.key() -> ""
     )
-    InteractiveSession.create(
-      0, "Test Interactive Session", null, None, livyConf, req, sessionStore, mockApp)
+    InteractiveSession.create(0, None, null, None, livyConf, req, sessionStore, mockApp)
   }
 
   private def executeStatement(code: String, codeType: Option[String] = None): JValue = {
@@ -159,6 +158,7 @@ class InteractiveSessionSpec extends FunSpec
       val mockApp = mock[SparkApp]
       val sessionStore = mock[SessionStore]
       session = createSession(sessionStore, Some(mockApp))
+      session.start()
 
       val expectedAppId = "APPID"
       session.appIdKnown(expectedAppId)
@@ -241,15 +241,32 @@ class InteractiveSessionSpec extends FunSpec
   }
 
   describe("recovery") {
-    it("should recover session") {
+    it("should recover named sessions") {
       val conf = new LivyConf()
       val sessionStore = mock[SessionStore]
       val mockClient = mock[RSCClient]
       when(mockClient.submit(any(classOf[PingJob]))).thenReturn(mock[JobHandle[Void]])
-      val m =
-        InteractiveRecoveryMetadata(
-          78, "Test session ", None, "appTag", Spark(), 0, null, None, Some(URI.create("")))
+      val m = InteractiveRecoveryMetadata(
+          78, Some("Test session"), None, "appTag", Spark(), 0, null, None, Some(URI.create("")))
+      val s = InteractiveSession.recover(m, conf, sessionStore, None, Some(mockClient))
+      s.start()
+
+      s.state shouldBe a[SessionState.Recovering]
+
+      s.appIdKnown("appId")
+      verify(sessionStore, atLeastOnce()).save(
+        MockitoMatchers.eq(InteractiveSession.RECOVERY_SESSION_TYPE), anyObject())
+    }
+
+    it("should recover sessions with no name") {
+      val conf = new LivyConf()
+      val sessionStore = mock[SessionStore]
+      val mockClient = mock[RSCClient]
+      when(mockClient.submit(any(classOf[PingJob]))).thenReturn(mock[JobHandle[Void]])
+      val m = InteractiveRecoveryMetadata(
+          78, None, None, "appTag", Spark(), 0, null, None, Some(URI.create("")))
       val s = InteractiveSession.recover(m, conf, sessionStore, None, Some(mockClient))
+      s.start()
 
       s.state shouldBe a[SessionState.Recovering]
 
@@ -262,9 +279,9 @@ class InteractiveSessionSpec extends FunSpec
       val conf = new LivyConf()
       val sessionStore = mock[SessionStore]
       val m = InteractiveRecoveryMetadata(
-        78, "Test session ", Some("appId"), "appTag", Spark(), 0, null, None, None)
+        78, None, Some("appId"), "appTag", Spark(), 0, null, None, None)
       val s = InteractiveSession.recover(m, conf, sessionStore, None)
-
+      s.start()
       s.state shouldBe a[SessionState.Dead]
       s.logLines().mkString should include("RSCDriver URI is unknown")
     }
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/SessionHeartbeatSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/SessionHeartbeatSpec.scala
index 86576b79e..6d2cff8af 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/SessionHeartbeatSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/SessionHeartbeatSpec.scala
@@ -56,7 +56,7 @@ class SessionHeartbeatSpec extends FunSpec with Matchers {
 
   describe("SessionHeartbeatWatchdog") {
     abstract class TestSession
-      extends Session(0, "Test Heart Beat Session", null, null) with SessionHeartbeat {}
+      extends Session(0, None, null, null) with SessionHeartbeat {}
     class TestWatchdog(conf: LivyConf)
       extends SessionManager[TestSession, RecoveryMetadata](
         conf,
@@ -69,12 +69,12 @@ class SessionHeartbeatSpec extends FunSpec with Matchers {
     it("should delete only expired sessions") {
       val expiredSession: TestSession = mock[TestSession]
       when(expiredSession.id).thenReturn(0)
-      when(expiredSession.name).thenReturn("expired session")
+      when(expiredSession.name).thenReturn(None)
       when(expiredSession.heartbeatExpired).thenReturn(true)
 
       val nonExpiredSession: TestSession = mock[TestSession]
       when(nonExpiredSession.id).thenReturn(1)
-      when(nonExpiredSession.name).thenReturn("unexpired session")
+      when(nonExpiredSession.name).thenReturn(None)
       when(nonExpiredSession.heartbeatExpired).thenReturn(false)
 
       val n = new TestWatchdog(new LivyConf())
diff --git a/server/src/test/scala/org/apache/livy/sessions/MockSession.scala b/server/src/test/scala/org/apache/livy/sessions/MockSession.scala
index 5e3c4665c..f848fb0de 100644
--- a/server/src/test/scala/org/apache/livy/sessions/MockSession.scala
+++ b/server/src/test/scala/org/apache/livy/sessions/MockSession.scala
@@ -19,12 +19,14 @@ package org.apache.livy.sessions
 
 import org.apache.livy.LivyConf
 
-class MockSession(id: Int, owner: String, conf: LivyConf, name: String = s"Mock-Session")
+class MockSession(id: Int, owner: String, conf: LivyConf, name: Option[String] = None)
   extends Session(id, name, owner, conf) {
   case class RecoveryMetadata(id: Int) extends Session.RecoveryMetadata()
 
   override val proxyUser = None
 
+  override def start(): Unit = ()
+
   override protected def stopSession(): Unit = ()
 
   override def logLines(): IndexedSeq[String] = IndexedSeq()
diff --git a/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala b/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
index 354d873ac..39acc2087 100644
--- a/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
+++ b/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
@@ -56,6 +56,7 @@ class SessionManagerSpec extends FunSpec with Matchers with LivyBaseUnitTestSuit
 
     it("should create sessions with names") {
       val livyConf = new LivyConf()
+      val name = "Mock-session"
       livyConf.set(LivyConf.SESSION_TIMEOUT, "100ms")
       val manager = new SessionManager[MockSession, RecoveryMetadata](
         livyConf,
@@ -63,13 +64,14 @@ class SessionManagerSpec extends FunSpec with Matchers with LivyBaseUnitTestSuit
         mock[SessionStore],
         "test",
         Some(Seq.empty))
-      val session = manager.register(new MockSession(manager.nextId(), null, livyConf, "session1"))
+      val session = manager.register(new MockSession(manager.nextId(), null, livyConf, Some(name)))
       manager.get(session.id).isDefined should be(true)
-      manager.get(session.name).isDefined should be(true)
+      manager.get(name).isDefined should be(true)
     }
 
     it("should not create sessions with the same name") {
       val livyConf = new LivyConf()
+      val name = "Mock-session"
       livyConf.set(LivyConf.SESSION_TIMEOUT, "100ms")
       val manager = new SessionManager[MockSession, RecoveryMetadata](
         livyConf,
@@ -77,8 +79,8 @@ class SessionManagerSpec extends FunSpec with Matchers with LivyBaseUnitTestSuit
         mock[SessionStore],
         "test",
         Some(Seq.empty))
-      val session1 = new MockSession(manager.nextId(), null, livyConf, "test session name")
-      val session2 = new MockSession(manager.nextId(), null, livyConf, "test session name")
+      val session1 = new MockSession(manager.nextId(), null, livyConf, Some(name))
+      val session2 = new MockSession(manager.nextId(), null, livyConf, Some(name))
       manager.register(session1)
       an[IllegalArgumentException] should be thrownBy manager.register(session2)
       manager.get(session1.id).isDefined should be(true)
@@ -147,7 +149,7 @@ class SessionManagerSpec extends FunSpec with Matchers with LivyBaseUnitTestSuit
     implicit def executor: ExecutionContext = ExecutionContext.global
 
     def makeMetadata(id: Int, appTag: String): BatchRecoveryMetadata = {
-      BatchRecoveryMetadata(id, s"test-session-$id", None, appTag, null, None)
+      BatchRecoveryMetadata(id, Some(s"test-session-$id"), None, appTag, null, None)
     }
 
     def mockSession(id: Int): BatchSession = {

From 0a0e1cf0f733b836461bb6add266836de03f5707 Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam(mfathisalmi)" <mfathisalmi@paypal.com>
Date: Mon, 16 Oct 2017 09:49:51 -0700
Subject: [PATCH 05/24] [LIVY-41] Making session names optional

If no session name is provided, do not use session names.
This makes it possible to recover sessions that were stored with no
name.
---
 .../livy/client/http/HttpClientSpec.scala     |  1 +
 .../apache/livy/server/SessionServlet.scala   |  1 -
 .../InteractiveSessionServlet.scala           |  6 ++--
 .../org/apache/livy/sessions/Session.scala    |  4 +--
 .../apache/livy/sessions/SessionManager.scala |  4 +--
 .../livy/server/batch/BatchServletSpec.scala  | 32 +++++++++++++++++--
 .../BaseInteractiveServletSpec.scala          |  1 +
 .../InteractiveSessionServletSpec.scala       |  1 +
 .../livy/sessions/SessionManagerSpec.scala    |  3 ++
 9 files changed, 41 insertions(+), 12 deletions(-)

diff --git a/client-http/src/test/scala/org/apache/livy/client/http/HttpClientSpec.scala b/client-http/src/test/scala/org/apache/livy/client/http/HttpClientSpec.scala
index 5ea6f8d40..8f38b450a 100644
--- a/client-http/src/test/scala/org/apache/livy/client/http/HttpClientSpec.scala
+++ b/client-http/src/test/scala/org/apache/livy/client/http/HttpClientSpec.scala
@@ -273,6 +273,7 @@ private class HttpClientTestBootstrap extends LifeCycle {
         val session = mock(classOf[InteractiveSession])
         val id = sessionManager.nextId()
         when(session.id).thenReturn(id)
+        when(session.name).thenReturn(None)
         when(session.appId).thenReturn(None)
         when(session.appInfo).thenReturn(AppInfo())
         when(session.state).thenReturn(SessionState.Idle())
diff --git a/server/src/main/scala/org/apache/livy/server/SessionServlet.scala b/server/src/main/scala/org/apache/livy/server/SessionServlet.scala
index c300aa3dc..d1bde6f27 100644
--- a/server/src/main/scala/org/apache/livy/server/SessionServlet.scala
+++ b/server/src/main/scala/org/apache/livy/server/SessionServlet.scala
@@ -22,7 +22,6 @@ import javax.servlet.http.HttpServletRequest
 import org.scalatra._
 import scala.concurrent._
 import scala.concurrent.duration._
-import scala.util.{Failure, Success, Try}
 
 import org.apache.livy.{LivyConf, Logging}
 import org.apache.livy.sessions.{Session, SessionManager}
diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala
index 761a13408..c5b871daa 100644
--- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala
+++ b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala
@@ -53,11 +53,9 @@ class InteractiveSessionServlet(
   override protected def createSession(req: HttpServletRequest): InteractiveSession = {
     val createRequest = bodyAs[CreateInteractiveRequest](req)
     val proxyUser = checkImpersonation(createRequest.proxyUser, req)
-    val sessionId: Int = sessionManager.nextId()
-    val sessionName = createRequest.name
     InteractiveSession.create(
-      sessionId,
-      sessionName,
+      sessionManager.nextId(),
+      createRequest.name,
       remoteUser(req),
       proxyUser,
       livyConf,
diff --git a/server/src/main/scala/org/apache/livy/sessions/Session.scala b/server/src/main/scala/org/apache/livy/sessions/Session.scala
index 563e2d494..7b7b2f96a 100644
--- a/server/src/main/scala/org/apache/livy/sessions/Session.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/Session.scala
@@ -28,9 +28,7 @@ import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.hadoop.fs.permission.FsPermission
 import org.apache.hadoop.security.UserGroupInformation
 
-import org.apache.livy.LivyConf
-import org.apache.livy.Logging
-import org.apache.livy.Utils
+import org.apache.livy.{LivyConf, Logging, Utils}
 import org.apache.livy.utils.AppInfo
 
 object Session {
diff --git a/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala b/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
index 82f96d894..c182b46ba 100644
--- a/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
@@ -104,9 +104,9 @@ class SessionManager[S <: Session, R <: RecoveryMetadata : ClassTag](
           // info(s"sessionNames = ${sessionsByName.keys.mkString}")
           sessionsByName.put(sessionName, session)
         }
-        sessions.put(session.id, session)
-        session.start()
       }
+      sessions.put(session.id, session)
+      session.start()
     }
     session
   }
diff --git a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
index 42ff6431e..79799ff39 100644
--- a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
@@ -122,7 +122,7 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
       jpost[Map[String, Any]]("/", createRequest, expectedStatus = SC_BAD_REQUEST) { _ => }
     }
 
-    it("should show session properties") {
+    it("should show session properties for sessions with a name") {
       val id = 0
       val name = "TEST-batch-session"
       val state = SessionState.Running()
@@ -144,7 +144,8 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
         .asInstanceOf[BatchSessionView]
 
       view.id shouldEqual id
-      view.name shouldEqual name
+      view.name.isDefined shouldBe true
+      view.name shouldEqual Some(name)
       view.state shouldEqual state.toString
       view.appId shouldEqual Some(appId)
       view.appInfo shouldEqual appInfo
@@ -152,4 +153,31 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
     }
   }
 
+  it("should show session properties for sessions without a name") {
+    val id = 0
+    val state = SessionState.Running()
+    val appId = "appid"
+    val appInfo = AppInfo(Some("DRIVER LOG URL"), Some("SPARK UI URL"))
+    val log = IndexedSeq[String]("log1", "log2")
+
+    val session = mock[BatchSession]
+    when(session.id).thenReturn(id)
+    when(session.name).thenReturn(None)
+    when(session.state).thenReturn(state)
+    when(session.appId).thenReturn(Some(appId))
+    when(session.appInfo).thenReturn(appInfo)
+    when(session.logLines()).thenReturn(log)
+
+    val req = mock[HttpServletRequest]
+
+    val view = servlet.asInstanceOf[BatchSessionServlet].clientSessionView(session, req)
+      .asInstanceOf[BatchSessionView]
+
+    view.id shouldEqual id
+    view.name.isDefined shouldBe false
+    view.state shouldEqual state.toString
+    view.appId shouldEqual Some(appId)
+    view.appInfo shouldEqual appInfo
+    view.log shouldEqual log
+  }
 }
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/BaseInteractiveServletSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/BaseInteractiveServletSpec.scala
index b401a9233..0c18a5f98 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/BaseInteractiveServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/BaseInteractiveServletSpec.scala
@@ -61,6 +61,7 @@ abstract class BaseInteractiveServletSpec
     val classpath = sys.props("java.class.path")
     val request = new CreateInteractiveRequest()
     request.kind = kind
+    request.name = None
     request.conf = extraConf ++ Map(
       RSCConf.Entry.LIVY_JARS.key() -> "",
       RSCConf.Entry.CLIENT_IN_PROCESS.key() -> inProcess.toString,
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
index 0fd5e8729..949093c2f 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
@@ -56,6 +56,7 @@ class InteractiveSessionServletSpec extends BaseInteractiveServletSpec {
 
       val session = mock[InteractiveSession]
       when(session.kind).thenReturn(Spark())
+      when(session.name).thenReturn(None)
       when(session.appId).thenReturn(None)
       when(session.appInfo).thenReturn(AppInfo())
       when(session.logLines()).thenReturn(IndexedSeq())
diff --git a/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala b/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
index 39acc2087..bc1d068d0 100644
--- a/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
+++ b/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
@@ -95,6 +95,7 @@ class SessionManagerSpec extends FunSpec with Matchers with LivyBaseUnitTestSuit
       val sessionId = 24
       val session = mock[BatchSession]
       when(session.id).thenReturn(sessionId)
+      when(session.name).thenReturn(None)
       when(session.stop()).thenReturn(Future {})
       when(session.lastActivity).thenReturn(System.nanoTime())
 
@@ -107,6 +108,7 @@ class SessionManagerSpec extends FunSpec with Matchers with LivyBaseUnitTestSuit
       val sessionId = 24
       val session = mock[InteractiveSession]
       when(session.id).thenReturn(sessionId)
+      when(session.name).thenReturn(None)
       when(session.stop()).thenReturn(Future {})
       when(session.lastActivity).thenReturn(System.nanoTime())
 
@@ -155,6 +157,7 @@ class SessionManagerSpec extends FunSpec with Matchers with LivyBaseUnitTestSuit
     def mockSession(id: Int): BatchSession = {
       val session = mock[BatchSession]
       when(session.id).thenReturn(id)
+      when(session.name).thenReturn(None)
       when(session.stop()).thenReturn(Future {})
       when(session.lastActivity).thenReturn(System.nanoTime())
 

From c16685a984f7a22090ed267e7d52c5191965c52f Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam(mfathisalmi)" <mfathisalmi@paypal.com>
Date: Mon, 14 Jan 2019 15:56:50 -0800
Subject: [PATCH 06/24] Change name from null to None in test cases

---
 .../src/test/scala/org/apache/livy/test/BatchIT.scala         | 4 ++--
 .../src/test/scala/org/apache/livy/test/InteractiveIT.scala   | 2 +-
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/integration-test/src/test/scala/org/apache/livy/test/BatchIT.scala b/integration-test/src/test/scala/org/apache/livy/test/BatchIT.scala
index acc96f8ae..11619298f 100644
--- a/integration-test/src/test/scala/org/apache/livy/test/BatchIT.scala
+++ b/integration-test/src/test/scala/org/apache/livy/test/BatchIT.scala
@@ -161,14 +161,14 @@ class BatchIT extends BaseIntegrationTestSuite with BeforeAndAfterAll {
   private def withScript[R]
     (scriptPath: String, args: List[String], sparkConf: Map[String, String] = Map.empty)
     (f: (LivyRestClient#BatchSession) => R): R = {
-    val s = livyClient.startBatch(null, scriptPath, None, args, sparkConf)
+    val s = livyClient.startBatch(None, scriptPath, None, args, sparkConf)
     withSession(s)(f)
   }
 
   private def withTestLib[R]
     (testClass: Class[_], args: List[String], sparkConf: Map[String, String] = Map.empty)
     (f: (LivyRestClient#BatchSession) => R): R = {
-    val s = livyClient.startBatch(null, testLibPath, Some(testClass.getName()), args, sparkConf)
+    val s = livyClient.startBatch(None, testLibPath, Some(testClass.getName()), args, sparkConf)
     withSession(s)(f)
   }
 }
diff --git a/integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala b/integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala
index 1225f3843..6fc50033e 100644
--- a/integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala
+++ b/integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala
@@ -191,7 +191,7 @@ class InteractiveIT extends BaseIntegrationTestSuite {
       waitForIdle: Boolean = true,
       heartbeatTimeoutInSecond: Int = 0)
     (f: (LivyRestClient#InteractiveSession) => R): R = {
-    withSession(livyClient.startSession(null, kind, sparkConf, heartbeatTimeoutInSecond)) { s =>
+    withSession(livyClient.startSession(None, kind, sparkConf, heartbeatTimeoutInSecond)) { s =>
       if (waitForIdle) {
         s.verifySessionIdle()
       }

From c34eca899cdc9c63ca018ba03c4836ee779a66fd Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam(mfathisalmi)" <mfathisalmi@paypal.com>
Date: Mon, 14 Jan 2019 15:57:12 -0800
Subject: [PATCH 07/24] Validate session names

Session names cannot be numbers.
---
 .../main/scala/org/apache/livy/sessions/Session.scala | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/server/src/main/scala/org/apache/livy/sessions/Session.scala b/server/src/main/scala/org/apache/livy/sessions/Session.scala
index 7b7b2f96a..c3605da88 100644
--- a/server/src/main/scala/org/apache/livy/sessions/Session.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/Session.scala
@@ -139,6 +139,17 @@ abstract class Session(val id: Int, val name: Option[String], val owner: String,
 
   protected implicit val executionContext = ExecutionContext.global
 
+  private def isValidName(name: String): Boolean = {
+    name.forall(_.isDigit)
+  }
+
+  name.filterNot(isValidName)
+    .foreach { _ =>
+      val errMsg = s"Session name cannot be a number: $name"
+      error(errMsg)
+      throw new IllegalArgumentException(errMsg)
+    }
+
   protected var _appId: Option[String] = None
 
   private var _lastActivity = System.nanoTime()

From c995ab20c4d5d3d4eda7174ad6b0dc955a8f04f8 Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam(mfathisalmi)" <mfathisalmi@paypal.com>
Date: Tue, 15 Jan 2019 12:44:38 -0800
Subject: [PATCH 08/24] Exclude numbers from valid session names.

A session name should have at least one non digit character in it.
---
 .../scala/org/apache/livy/sessions/Session.scala  | 15 +++++++--------
 1 file changed, 7 insertions(+), 8 deletions(-)

diff --git a/server/src/main/scala/org/apache/livy/sessions/Session.scala b/server/src/main/scala/org/apache/livy/sessions/Session.scala
index a4fb2097c..749d24369 100644
--- a/server/src/main/scala/org/apache/livy/sessions/Session.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/Session.scala
@@ -142,16 +142,15 @@ abstract class Session(val id: Int, val name: Option[String], val owner: String,
 
   protected implicit val executionContext = ExecutionContext.global
 
-  private def isValidName(name: String): Boolean = {
-    name.forall(_.isDigit)
+  private def isSessionId(str: String): Boolean = {
+    str.nonEmpty && str.forall(_.isDigit)
   }
 
-  name.filterNot(isValidName)
-    .foreach { _ =>
-      val errMsg = s"Session name cannot be a number: $name"
-      error(errMsg)
-      throw new IllegalArgumentException(errMsg)
-    }
+  name.filter(isSessionId).foreach { idStr =>
+    val errMsg = s"Session name cannot be a number: $idStr"
+    error(errMsg)
+    throw new IllegalArgumentException(errMsg)
+  }
 
   protected var _appId: Option[String] = None
 

From 9f3915d237b9eee5262730be6e1d4a45d5db7d4d Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam(mfathisalmi)" <mfathisalmi@paypal.com>
Date: Tue, 15 Jan 2019 13:49:07 -0800
Subject: [PATCH 09/24] Cosmetics and code formatting

---
 .../apache/livy/server/interactive/InteractiveSession.scala    | 3 ++-
 .../main/scala/org/apache/livy/sessions/SessionManager.scala   | 2 +-
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
index d6228c53a..16c02d211 100644
--- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
@@ -422,7 +422,8 @@ class InteractiveSession(
 
         override def onJobSucceeded(job: JobHandle[Void], result: Void): Unit = {
           transition(SessionState.Running)
-          info(s"Interactive session $id created [appid: ${appId.orNull}, owner: $owner, proxyUser:" +
+          info(s"Interactive session $id created [appid: ${appId.orNull}, " +
+            s"owner: $owner, proxyUser:" +
             s" $proxyUser, state: ${state.toString}, kind: ${kind.toString}, " +
             s"info: ${appInfo.asJavaMap}]")
         }
diff --git a/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala b/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
index bc6f0862c..13a0a32b1 100644
--- a/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
@@ -94,7 +94,7 @@ class SessionManager[S <: Session, R <: RecoveryMetadata : ClassTag](
   def register(session: S): S = {
     info(s"Registering new session ${session.id}")
     synchronized {
-      session.name.map { sessionName =>
+      session.name.foreach { sessionName =>
         if (sessionsByName.contains(sessionName)) {
           val msg = s"Session ${session.name} already exists!"
           error(msg)

From ecde1fdc61c0503e67caae685267fb817c8864f1 Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam(mfathisalmi)" <mfathisalmi@paypal.com>
Date: Tue, 15 Jan 2019 14:14:47 -0800
Subject: [PATCH 10/24] Set Thriftserver session names to None

---
 .../org/apache/livy/thriftserver/LivyThriftSessionManager.scala  | 1 +
 1 file changed, 1 insertion(+)

diff --git a/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/LivyThriftSessionManager.scala b/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/LivyThriftSessionManager.scala
index 5be5536d7..31ea2f0ab 100644
--- a/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/LivyThriftSessionManager.scala
+++ b/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/LivyThriftSessionManager.scala
@@ -228,6 +228,7 @@ class LivyThriftSessionManager(val server: LivyThriftServer, val livyConf: LivyC
       createInteractiveRequest.kind = Spark
       val newSession = InteractiveSession.create(
         server.livySessionManager.nextId(),
+        None,
         username,
         server.livyConf,
         server.accessManager,

From 57ec81abacaacf7bbf76d544c45aa3d97f17e7e4 Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam(mfathisalmi)" <mfathisalmi@paypal.com>
Date: Tue, 15 Jan 2019 15:06:50 -0800
Subject: [PATCH 11/24] change method calls from infix form to . form in
 InteractiveSession

---
 .../apache/livy/server/interactive/InteractiveSession.scala   | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
index 16c02d211..5e80b58e3 100644
--- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
@@ -402,13 +402,13 @@ class InteractiveSession(
     } else {
       val uriFuture = Future { client.get.getServerUri.get() }
 
-      uriFuture onSuccess { case url =>
+      uriFuture.onSuccess { case url =>
         rscDriverUri = Option(url)
         sessionSaveLock.synchronized {
           sessionStore.save(RECOVERY_SESSION_TYPE, recoveryMetadata)
         }
       }
-      uriFuture onFailure { case e => warn("Fail to get rsc uri", e) }
+      uriFuture.onFailure { case e => warn("Fail to get rsc uri", e) }
 
       // Send a dummy job that will return once the client is ready to be used, and set the
       // state to "idle" at that point.

From 2065619a9500eed2c5669900b71525a19fd8a369 Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam(mfathisalmi)" <mfathisalmi@paypal.com>
Date: Tue, 22 Jan 2019 10:18:52 -0800
Subject: [PATCH 12/24] Test sessions without a name can be recovered.

After Livy-41, sessions can have an optional name.
If a Livy server is updated to include Livy-41,
it should recover previous sessions that do not have a name.
---
 .../recovery/BlackholeStateStoreSpec.scala    | 20 +++++++++++++++++++
 1 file changed, 20 insertions(+)

diff --git a/server/src/test/scala/org/apache/livy/server/recovery/BlackholeStateStoreSpec.scala b/server/src/test/scala/org/apache/livy/server/recovery/BlackholeStateStoreSpec.scala
index e40bb1c03..477bce83a 100644
--- a/server/src/test/scala/org/apache/livy/server/recovery/BlackholeStateStoreSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/recovery/BlackholeStateStoreSpec.scala
@@ -21,6 +21,7 @@ import org.scalatest.FunSpec
 import org.scalatest.Matchers._
 
 import org.apache.livy.{LivyBaseUnitTestSuite, LivyConf}
+import org.apache.livy.server.batch.BatchRecoveryMetadata
 
 class BlackholeStateStoreSpec extends FunSpec with LivyBaseUnitTestSuite {
   describe("BlackholeStateStore") {
@@ -43,5 +44,24 @@ class BlackholeStateStoreSpec extends FunSpec with LivyBaseUnitTestSuite {
     it("remove should not throw") {
       stateStore.remove("")
     }
+
+    it("should deserialize sessions without name") {
+      val json =
+        """
+          |{
+          |  "id": 408107,
+          |  "appId": "application_1541532370353_1465148",
+          |  "state": "running",
+          |  "appTag": "livy-batch-408107-2jAOFzDy",
+          |  "owner": "batch_admin",
+          |  "proxyUser": "batch_opts",
+          |  "version": 1
+          |}
+        """.stripMargin
+      val batchRecoveryMetadata = stateStore.deserialize[BatchRecoveryMetadata](json.getBytes())
+      batchRecoveryMetadata.id shouldBe 408107
+      batchRecoveryMetadata.appId shouldBe Some("application_1541532370353_1465148")
+      batchRecoveryMetadata.name shouldBe None
+    }
   }
 }

From 962e82e033ca3b0a92e6235a2789db927a82bca0 Mon Sep 17 00:00:00 2001
From: Meisam Fathi <meisam.fathi@gmail.com>
Date: Fri, 25 Jan 2019 17:25:02 -0800
Subject: [PATCH 13/24] Cosmetics and code formatting

---
 .../scala/org/apache/livy/server/batch/BatchSession.scala  | 3 +--
 .../src/main/scala/org/apache/livy/sessions/Session.scala  | 7 +++++--
 .../scala/org/apache/livy/sessions/SessionManager.scala    | 1 -
 3 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala b/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
index a63746c0f..3bbd7425c 100644
--- a/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
@@ -174,8 +174,7 @@ class BatchSession(
   }
 
   override def appIdKnown(appId: String): Unit = {
-    _appId = Option(
-      appId)
+    _appId = Option(appId)
     sessionStore.save(RECOVERY_SESSION_TYPE, recoveryMetadata)
   }
 
diff --git a/server/src/main/scala/org/apache/livy/sessions/Session.scala b/server/src/main/scala/org/apache/livy/sessions/Session.scala
index 749d24369..a8399c6c4 100644
--- a/server/src/main/scala/org/apache/livy/sessions/Session.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/Session.scala
@@ -134,8 +134,11 @@ object Session {
   }
 }
 
-abstract class Session(val id: Int, val name: Option[String], val owner: String,
-                       val livyConf: LivyConf)
+abstract class Session(
+    val id: Int,
+    val name: Option[String],
+    val owner: String,
+    val livyConf: LivyConf)
   extends Logging {
 
   import Session._
diff --git a/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala b/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
index 13a0a32b1..4875d5065 100644
--- a/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
@@ -101,7 +101,6 @@ class SessionManager[S <: Session, R <: RecoveryMetadata : ClassTag](
           delete(session)
           throw new IllegalArgumentException(msg)
         } else {
-          // info(s"sessionNames = ${sessionsByName.keys.mkString}")
           sessionsByName.put(sessionName, session)
         }
       }

From 3527ef700bf704e60d0bc3a26b09fbf4878c179c Mon Sep 17 00:00:00 2001
From: Meisam Fathi <meisam.fathi@gmail.com>
Date: Fri, 25 Jan 2019 17:26:03 -0800
Subject: [PATCH 14/24] Simplify check for session names.

---
 .../src/main/scala/org/apache/livy/sessions/Session.scala   | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/server/src/main/scala/org/apache/livy/sessions/Session.scala b/server/src/main/scala/org/apache/livy/sessions/Session.scala
index a8399c6c4..265e81f18 100644
--- a/server/src/main/scala/org/apache/livy/sessions/Session.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/Session.scala
@@ -149,10 +149,8 @@ abstract class Session(
     str.nonEmpty && str.forall(_.isDigit)
   }
 
-  name.filter(isSessionId).foreach { idStr =>
-    val errMsg = s"Session name cannot be a number: $idStr"
-    error(errMsg)
-    throw new IllegalArgumentException(errMsg)
+  name.foreach { idStr =>
+    if (isSessionId(idStr)) throw new IllegalArgumentException(s"Invalid session name: $idStr")
   }
 
   protected var _appId: Option[String] = None

From 9c64c818a1d0bc8ddfab52bf506c8cd558a7fb4c Mon Sep 17 00:00:00 2001
From: Meisam Fathi <meisam.fathi@gmail.com>
Date: Fri, 25 Jan 2019 17:27:23 -0800
Subject: [PATCH 15/24] Start heartbeat and save session recovery data after
 start() is called.

---
 .../apache/livy/server/interactive/InteractiveSession.scala  | 4 ++--
 .../main/scala/org/apache/livy/sessions/SessionManager.scala | 5 +----
 2 files changed, 3 insertions(+), 6 deletions(-)

diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
index 5e80b58e3..0c3a3a82e 100644
--- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
@@ -382,12 +382,12 @@ class InteractiveSession(
   private val sessionSaveLock = new Object()
 
   _appId = appIdHint
-  sessionStore.save(RECOVERY_SESSION_TYPE, recoveryMetadata)
-  heartbeat()
 
   private var app: Option[SparkApp] = None
 
   override def start(): Unit = {
+    sessionStore.save(RECOVERY_SESSION_TYPE, recoveryMetadata)
+    heartbeat()
     app = mockApp.orElse {
       val driverProcess = client.flatMap { c => Option(c.getDriverProcess) }
         .map(new LineBufferedProcess(_, livyConf.getInt(LivyConf.SPARK_LOGS_SIZE)))
diff --git a/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala b/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
index 4875d5065..a63cab3e4 100644
--- a/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/SessionManager.scala
@@ -96,10 +96,7 @@ class SessionManager[S <: Session, R <: RecoveryMetadata : ClassTag](
     synchronized {
       session.name.foreach { sessionName =>
         if (sessionsByName.contains(sessionName)) {
-          val msg = s"Session ${session.name} already exists!"
-          error(msg)
-          delete(session)
-          throw new IllegalArgumentException(msg)
+          throw new IllegalArgumentException(s"Duplicate session name: ${session.name}")
         } else {
           sessionsByName.put(sessionName, session)
         }

From 516a8805f357a76068c560506df6bae8a15ba9f0 Mon Sep 17 00:00:00 2001
From: Meisam Fathi <meisam.fathi@gmail.com>
Date: Tue, 29 Jan 2019 16:22:02 -0800
Subject: [PATCH 16/24] Inline check for session names.

Also make it reject empty names
---
 .../main/scala/org/apache/livy/sessions/Session.scala  | 10 ++++------
 1 file changed, 4 insertions(+), 6 deletions(-)

diff --git a/server/src/main/scala/org/apache/livy/sessions/Session.scala b/server/src/main/scala/org/apache/livy/sessions/Session.scala
index 265e81f18..3e3830f6a 100644
--- a/server/src/main/scala/org/apache/livy/sessions/Session.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/Session.scala
@@ -145,12 +145,10 @@ abstract class Session(
 
   protected implicit val executionContext = ExecutionContext.global
 
-  private def isSessionId(str: String): Boolean = {
-    str.nonEmpty && str.forall(_.isDigit)
-  }
-
-  name.foreach { idStr =>
-    if (isSessionId(idStr)) throw new IllegalArgumentException(s"Invalid session name: $idStr")
+  // validate session name. The name should not be a number
+  name.foreach { sessionName =>
+    if (sessionName.forall(_.isDigit))
+      throw new IllegalArgumentException(s"Invalid session name: $sessionName")
   }
 
   protected var _appId: Option[String] = None

From 334b165d0f75691f4c7ae82608c89f4b12703d97 Mon Sep 17 00:00:00 2001
From: Meisam Fathi <meisam.fathi@gmail.com>
Date: Tue, 29 Jan 2019 16:24:52 -0800
Subject: [PATCH 17/24] Refactor common test case code snippets into helper
 functions

---
 .../livy/server/batch/BatchServletSpec.scala  | 91 +++++++------------
 .../livy/server/batch/BatchSessionSpec.scala  | 25 ++---
 .../InteractiveSessionServletSpec.scala       | 50 ++--------
 .../livy/sessions/SessionManagerSpec.scala    | 41 ++++-----
 4 files changed, 69 insertions(+), 138 deletions(-)

diff --git a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
index 6738220ae..874c4fe19 100644
--- a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
@@ -62,6 +62,34 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
       accessManager)
   }
 
+  def testShowSessionProperties(id: Int, name: Option[String]): Unit = {
+    val state = SessionState.Running
+    val appId = "appid"
+    val appInfo = AppInfo(Some("DRIVER LOG URL"), Some("SPARK UI URL"))
+    val log = IndexedSeq[String]("log1", "log2")
+
+    val session = mock[BatchSession]
+    when(session.id).thenReturn(id)
+    when(session.name).thenReturn(name)
+    when(session.state).thenReturn(state)
+    when(session.appId).thenReturn(Some(appId))
+    when(session.appInfo).thenReturn(appInfo)
+    when(session.logLines()).thenReturn(log)
+
+    val req = mock[HttpServletRequest]
+
+    val view = servlet.asInstanceOf[BatchSessionServlet].clientSessionView(session, req)
+      .asInstanceOf[BatchSessionView]
+
+    view.id shouldEqual id
+    view.name.isDefined shouldBe true
+    view.name shouldEqual Some(name)
+    view.state shouldEqual state.toString
+    view.appId shouldEqual Some(appId)
+    view.appInfo shouldEqual appInfo
+    view.log shouldEqual log
+  }
+
   describe("Batch Servlet") {
     it("should create and tear down a batch") {
       jget[Map[String, Any]]("/") { data =>
@@ -122,63 +150,12 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
       jpost[Map[String, Any]]("/", createRequest, expectedStatus = SC_BAD_REQUEST) { _ => }
     }
 
-    it("should show session properties for sessions with a name") {
-      val id = 0
-      val name = "TEST-batch-session"
-      val state = SessionState.Running
-      val appId = "appid"
-      val appInfo = AppInfo(Some("DRIVER LOG URL"), Some("SPARK UI URL"))
-      val log = IndexedSeq[String]("log1", "log2")
-
-      val session = mock[BatchSession]
-      when(session.id).thenReturn(id)
-      when(session.name).thenReturn(Some(name))
-      when(session.state).thenReturn(state)
-      when(session.appId).thenReturn(Some(appId))
-      when(session.appInfo).thenReturn(appInfo)
-      when(session.logLines()).thenReturn(log)
-
-      val req = mock[HttpServletRequest]
-
-      val view = servlet.asInstanceOf[BatchSessionServlet].clientSessionView(session, req)
-        .asInstanceOf[BatchSessionView]
-
-      view.id shouldEqual id
-      view.name.isDefined shouldBe true
-      view.name shouldEqual Some(name)
-      view.state shouldEqual state.toString
-      view.appId shouldEqual Some(appId)
-      view.appInfo shouldEqual appInfo
-      view.log shouldEqual log
-    }
-
-    it("should show session properties for sessions without a name") {
-      val id = 0
-      val state = SessionState.Running
-      val appId = "appid"
-      val appInfo = AppInfo(Some("DRIVER LOG URL"), Some("SPARK UI URL"))
-      val log = IndexedSeq[String]("log1", "log2")
-
-      val session = mock[BatchSession]
-      when(session.id).thenReturn(id)
-      when(session.name).thenReturn(None)
-      when(session.state).thenReturn(state)
-      when(session.appId).thenReturn(Some(appId))
-      when(session.appInfo).thenReturn(appInfo)
-      when(session.logLines()).thenReturn(log)
-
-      val req = mock[HttpServletRequest]
-
-      val view = servlet.asInstanceOf[BatchSessionServlet].clientSessionView(session, req)
-        .asInstanceOf[BatchSessionView]
-
-      view.id shouldEqual id
-      view.name.isDefined shouldBe false
-      view.state shouldEqual state.toString
-      view.appId shouldEqual Some(appId)
-      view.appInfo shouldEqual appInfo
-      view.log shouldEqual log
-    }
+    Seq((0, None), (0, Some("TEST-batch-session")))
+      .foreach { case (id, name) =>
+        it(s"should show session properties (id = $id, name = $name)") {
+          testShowSessionProperties(id, name)
+        }
+      }
 
     it("should fail session creation when max session creation is hit") {
       val createRequest = new CreateBatchRequest()
diff --git a/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala b/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala
index aa6e0fcaa..d7cac1578 100644
--- a/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/batch/BatchSessionSpec.scala
@@ -102,32 +102,27 @@ class BatchSessionSpec
       batch.appInfo shouldEqual expectedAppInfo
     }
 
-    it("should recover named session") {
+    def testRecoverSession(name: Option[String]): Unit = {
       val conf = new LivyConf()
       val req = new CreateBatchRequest()
+      val name = Some("Test Batch Session")
       val mockApp = mock[SparkApp]
-      val m = BatchRecoveryMetadata(99, Some("Test Batch Session"), None, "appTag", null, None)
+      val m = BatchRecoveryMetadata(99, name, None, "appTag", null, None)
       val batch = BatchSession.recover(m, conf, sessionStore, Some(mockApp))
 
       batch.state shouldBe (SessionState.Recovering)
+      batch.name shouldBe (name)
 
       batch.appIdKnown("appId")
       verify(sessionStore, atLeastOnce()).save(
         Matchers.eq(BatchSession.RECOVERY_SESSION_TYPE), anyObject())
     }
 
-    it("should recover session with no name ") {
-      val conf = new LivyConf()
-      val req = new CreateBatchRequest()
-      val mockApp = mock[SparkApp]
-      val m = BatchRecoveryMetadata(999, None, None, "appTag", null, None)
-      val batch = BatchSession.recover(m, conf, sessionStore, Some(mockApp))
-
-      batch.state shouldBe (SessionState.Recovering)
-
-      batch.appIdKnown("appId")
-      verify(sessionStore, atLeastOnce()).save(
-        Matchers.eq(BatchSession.RECOVERY_SESSION_TYPE), anyObject())
-    }
+    Seq[Option[String]](None, Some("Test Batch Session"), null)
+      .foreach { case name =>
+        it(s"should recover session (name = $name)") {
+          testRecoverSession(name)
+        }
+      }
   }
 }
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
index 2c296de84..792a11770 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
@@ -152,46 +152,14 @@ class InteractiveSessionServletSpec extends BaseInteractiveServletSpec {
     }
   }
 
-  it("should show session properties") {
-    val id = 0
-    val name = "TEST-interactive-session"
-    val appId = "appid"
-    val owner = "owner"
-    val proxyUser = "proxyUser"
-    val state = SessionState.Running
-    val kind = Spark
-    val appInfo = AppInfo(Some("DRIVER LOG URL"), Some("SPARK UI URL"))
-    val log = IndexedSeq[String]("log1", "log2")
-
-    val session = mock[InteractiveSession]
-    when(session.id).thenReturn(id)
-    when(session.name).thenReturn(Some(name))
-    when(session.appId).thenReturn(Some(appId))
-    when(session.owner).thenReturn(owner)
-    when(session.proxyUser).thenReturn(Some(proxyUser))
-    when(session.state).thenReturn(state)
-    when(session.kind).thenReturn(kind)
-    when(session.appInfo).thenReturn(appInfo)
-    when(session.logLines()).thenReturn(log)
-
-    val req = mock[HttpServletRequest]
-
-    val view = servlet.asInstanceOf[InteractiveSessionServlet].clientSessionView(session, req)
-      .asInstanceOf[SessionInfo]
-
-    view.id shouldEqual id
-    view.name shouldEqual name
-    view.appId shouldEqual appId
-    view.owner shouldEqual owner
-    view.proxyUser shouldEqual proxyUser
-    view.state shouldEqual state.toString
-    view.kind shouldEqual kind.toString
-    view.appInfo should contain (Entry(AppInfo.DRIVER_LOG_URL_NAME, appInfo.driverLogUrl.get))
-    view.appInfo should contain (Entry(AppInfo.SPARK_UI_URL_NAME, appInfo.sparkUiUrl.get))
-    view.log shouldEqual log.asJava
-  }
+  Seq(Some("TEST-interactive-session"), None, null)
+    .foreach { case name =>
+      it(s"should show session properties (name =$name") {
+        testShowSessionProperties(name: Option[String])
+      }
+    }
 
-  it("should show session properties for sessions without a name") {
+  def testShowSessionProperties(name: Option[String]): Unit = {
     val id = 0
     val appId = "appid"
     val owner = "owner"
@@ -203,7 +171,7 @@ class InteractiveSessionServletSpec extends BaseInteractiveServletSpec {
 
     val session = mock[InteractiveSession]
     when(session.id).thenReturn(id)
-    when(session.name).thenReturn(None)
+    when(session.name).thenReturn(name)
     when(session.appId).thenReturn(Some(appId))
     when(session.owner).thenReturn(owner)
     when(session.proxyUser).thenReturn(Some(proxyUser))
@@ -218,7 +186,7 @@ class InteractiveSessionServletSpec extends BaseInteractiveServletSpec {
       .asInstanceOf[SessionInfo]
 
     view.id shouldEqual id
-    view.name shouldEqual null
+    view.name shouldEqual name
     view.appId shouldEqual appId
     view.owner shouldEqual owner
     view.proxyUser shouldEqual proxyUser
diff --git a/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala b/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
index a5ffb0841..a1d6b33be 100644
--- a/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
+++ b/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
@@ -36,16 +36,21 @@ import org.apache.livy.sessions.Session.RecoveryMetadata
 class SessionManagerSpec extends FunSpec with Matchers with LivyBaseUnitTestSuite {
   implicit def executor: ExecutionContext = ExecutionContext.global
 
+  private def createSessionManager(): (LivyConf, SessionManager[MockSession, RecoveryMetadata]) = {
+    val livyConf = new LivyConf()
+    livyConf.set(LivyConf.SESSION_TIMEOUT, "100ms")
+    val manager = new SessionManager[MockSession, RecoveryMetadata](
+      livyConf,
+      { _ => assert(false).asInstanceOf[MockSession] },
+      mock[SessionStore],
+      "test",
+      Some(Seq.empty))
+    (livyConf, manager)
+  }
+
   describe("SessionManager") {
     it("should garbage collect old sessions") {
-      val livyConf = new LivyConf()
-      livyConf.set(LivyConf.SESSION_TIMEOUT, "100ms")
-      val manager = new SessionManager[MockSession, RecoveryMetadata](
-        livyConf,
-        { _ => assert(false).asInstanceOf[MockSession] },
-        mock[SessionStore],
-        "test",
-        Some(Seq.empty))
+      val (livyConf, manager) = createSessionManager()
       val session = manager.register(new MockSession(manager.nextId(), null, livyConf))
       manager.get(session.id).isDefined should be(true)
       eventually(timeout(5 seconds), interval(100 millis)) {
@@ -55,30 +60,16 @@ class SessionManagerSpec extends FunSpec with Matchers with LivyBaseUnitTestSuit
     }
 
     it("should create sessions with names") {
-      val livyConf = new LivyConf()
+      val (livyConf, manager) = createSessionManager()
       val name = "Mock-session"
-      livyConf.set(LivyConf.SESSION_TIMEOUT, "100ms")
-      val manager = new SessionManager[MockSession, RecoveryMetadata](
-        livyConf,
-        { _ => assert(false).asInstanceOf[MockSession] },
-        mock[SessionStore],
-        "test",
-        Some(Seq.empty))
       val session = manager.register(new MockSession(manager.nextId(), null, livyConf, Some(name)))
       manager.get(session.id).isDefined should be(true)
       manager.get(name).isDefined should be(true)
     }
 
-    it("should not create sessions with the same name") {
-      val livyConf = new LivyConf()
+    it("should not create sessions with duplicate names") {
+      val (livyConf, manager) = createSessionManager()
       val name = "Mock-session"
-      livyConf.set(LivyConf.SESSION_TIMEOUT, "100ms")
-      val manager = new SessionManager[MockSession, RecoveryMetadata](
-        livyConf,
-        { _ => assert(false).asInstanceOf[MockSession] },
-        mock[SessionStore],
-        "test",
-        Some(Seq.empty))
       val session1 = new MockSession(manager.nextId(), null, livyConf, Some(name))
       val session2 = new MockSession(manager.nextId(), null, livyConf, Some(name))
       manager.register(session1)

From ef56dead13d2775e8eebf07f53e70d2a8a5db466 Mon Sep 17 00:00:00 2001
From: Meisam Fathi <meisam.fathi@gmail.com>
Date: Tue, 29 Jan 2019 16:25:39 -0800
Subject: [PATCH 18/24] Decode string bytes as "UTF-8"

---
 .../livy/server/recovery/BlackholeStateStoreSpec.scala      | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/server/src/test/scala/org/apache/livy/server/recovery/BlackholeStateStoreSpec.scala b/server/src/test/scala/org/apache/livy/server/recovery/BlackholeStateStoreSpec.scala
index 477bce83a..7b361e645 100644
--- a/server/src/test/scala/org/apache/livy/server/recovery/BlackholeStateStoreSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/recovery/BlackholeStateStoreSpec.scala
@@ -46,7 +46,7 @@ class BlackholeStateStoreSpec extends FunSpec with LivyBaseUnitTestSuite {
     }
 
     it("should deserialize sessions without name") {
-      val json =
+      val jsonbytes =
         """
           |{
           |  "id": 408107,
@@ -57,8 +57,8 @@ class BlackholeStateStoreSpec extends FunSpec with LivyBaseUnitTestSuite {
           |  "proxyUser": "batch_opts",
           |  "version": 1
           |}
-        """.stripMargin
-      val batchRecoveryMetadata = stateStore.deserialize[BatchRecoveryMetadata](json.getBytes())
+        """.stripMargin.getBytes("UTF_8")
+      val batchRecoveryMetadata = stateStore.deserialize[BatchRecoveryMetadata](jsonbytes)
       batchRecoveryMetadata.id shouldBe 408107
       batchRecoveryMetadata.appId shouldBe Some("application_1541532370353_1465148")
       batchRecoveryMetadata.name shouldBe None

From 41ff9aa06aac13563b3ade9f92e3e3da20c7b85f Mon Sep 17 00:00:00 2001
From: Meisam Fathi <meisam.fathi@gmail.com>
Date: Tue, 29 Jan 2019 16:26:11 -0800
Subject: [PATCH 19/24] Remove the irrelevant checks for the duplicate session
 names test

---
 .../scala/org/apache/livy/sessions/SessionManagerSpec.scala  | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala b/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
index a1d6b33be..523e1d7b2 100644
--- a/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
+++ b/server/src/test/scala/org/apache/livy/sessions/SessionManagerSpec.scala
@@ -76,10 +76,7 @@ class SessionManagerSpec extends FunSpec with Matchers with LivyBaseUnitTestSuit
       an[IllegalArgumentException] should be thrownBy manager.register(session2)
       manager.get(session1.id).isDefined should be(true)
       manager.get(session2.id).isDefined should be(false)
-      eventually(timeout(5 seconds), interval(100 millis)) {
-        Await.result(manager.collectGarbage(), Duration.Inf)
-        manager.get(session1.id) should be(None)
-      }
+      manager.shutdown()
     }
 
     it("batch session should not be gc-ed until application is finished") {

From d673baffb2eb7c3310ff9066776125f928a3123d Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam" <meisam.fathi@gmail.com>
Date: Tue, 29 Jan 2019 18:19:12 -0800
Subject: [PATCH 20/24] Guard if caluse with braces

---
 server/src/main/scala/org/apache/livy/sessions/Session.scala | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/server/src/main/scala/org/apache/livy/sessions/Session.scala b/server/src/main/scala/org/apache/livy/sessions/Session.scala
index 3e3830f6a..ee142835a 100644
--- a/server/src/main/scala/org/apache/livy/sessions/Session.scala
+++ b/server/src/main/scala/org/apache/livy/sessions/Session.scala
@@ -147,8 +147,9 @@ abstract class Session(
 
   // validate session name. The name should not be a number
   name.foreach { sessionName =>
-    if (sessionName.forall(_.isDigit))
+    if (sessionName.forall(_.isDigit)) {
       throw new IllegalArgumentException(s"Invalid session name: $sessionName")
+    }
   }
 
   protected var _appId: Option[String] = None

From a43a968e922a631454d1bcc65ef736bf1aa26124 Mon Sep 17 00:00:00 2001
From: "Fathi, Meisam" <meisam.fathi@gmail.com>
Date: Wed, 30 Jan 2019 07:16:16 -0800
Subject: [PATCH 21/24] [LIVY-41] Fix typo : "UTF_8" -> "UTF-8"

---
 .../apache/livy/server/recovery/BlackholeStateStoreSpec.scala   | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/server/src/test/scala/org/apache/livy/server/recovery/BlackholeStateStoreSpec.scala b/server/src/test/scala/org/apache/livy/server/recovery/BlackholeStateStoreSpec.scala
index 7b361e645..8ee448f5e 100644
--- a/server/src/test/scala/org/apache/livy/server/recovery/BlackholeStateStoreSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/recovery/BlackholeStateStoreSpec.scala
@@ -57,7 +57,7 @@ class BlackholeStateStoreSpec extends FunSpec with LivyBaseUnitTestSuite {
           |  "proxyUser": "batch_opts",
           |  "version": 1
           |}
-        """.stripMargin.getBytes("UTF_8")
+        """.stripMargin.getBytes("UTF-8")
       val batchRecoveryMetadata = stateStore.deserialize[BatchRecoveryMetadata](jsonbytes)
       batchRecoveryMetadata.id shouldBe 408107
       batchRecoveryMetadata.appId shouldBe Some("application_1541532370353_1465148")

From 35c34d730d4e6a5636a8dfdff6df110d459c3451 Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam" <meisam.fathi@gmail.com>
Date: Wed, 30 Jan 2019 08:18:56 -0800
Subject: [PATCH 22/24] [LIVY-41] Fix tests for session names

---
 .../scala/org/apache/livy/server/batch/BatchServletSpec.scala  | 3 +--
 .../server/interactive/InteractiveSessionServletSpec.scala     | 2 +-
 2 files changed, 2 insertions(+), 3 deletions(-)

diff --git a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
index 874c4fe19..1ff347663 100644
--- a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
@@ -82,8 +82,7 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
       .asInstanceOf[BatchSessionView]
 
     view.id shouldEqual id
-    view.name.isDefined shouldBe true
-    view.name shouldEqual Some(name)
+    Option(view.name) shouldEqual name
     view.state shouldEqual state.toString
     view.appId shouldEqual Some(appId)
     view.appInfo shouldEqual appInfo
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
index 792a11770..6c97debfe 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
@@ -186,7 +186,7 @@ class InteractiveSessionServletSpec extends BaseInteractiveServletSpec {
       .asInstanceOf[SessionInfo]
 
     view.id shouldEqual id
-    view.name shouldEqual name
+    Option(view.name) shouldEqual name
     view.appId shouldEqual appId
     view.owner shouldEqual owner
     view.proxyUser shouldEqual proxyUser

From 518bfdf28b8e84bafc4168ceebcf49fa8ebbd115 Mon Sep 17 00:00:00 2001
From: "Fathi Salmi, Meisam" <meisam.fathi@gmail.com>
Date: Wed, 30 Jan 2019 08:19:30 -0800
Subject: [PATCH 23/24] [LIVY-41] Remove ID parameter from
 testShowSessionProperties

---
 .../apache/livy/server/batch/BatchServletSpec.scala   | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
index 1ff347663..505a61698 100644
--- a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
@@ -62,7 +62,8 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
       accessManager)
   }
 
-  def testShowSessionProperties(id: Int, name: Option[String]): Unit = {
+  def testShowSessionProperties(name: Option[String]): Unit = {
+    val id = 0
     val state = SessionState.Running
     val appId = "appid"
     val appInfo = AppInfo(Some("DRIVER LOG URL"), Some("SPARK UI URL"))
@@ -149,10 +150,10 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
       jpost[Map[String, Any]]("/", createRequest, expectedStatus = SC_BAD_REQUEST) { _ => }
     }
 
-    Seq((0, None), (0, Some("TEST-batch-session")))
-      .foreach { case (id, name) =>
-        it(s"should show session properties (id = $id, name = $name)") {
-          testShowSessionProperties(id, name)
+    Seq(None, Some("TEST-batch-session"))
+      .foreach { name =>
+        it(s"should show session properties (name = $name)") {
+          testShowSessionProperties(name)
         }
       }
 

From f61c3d5bcf790ca093640c1f6396cf2b8de9986c Mon Sep 17 00:00:00 2001
From: Meisam Fathi <meisam.fathi@gmail.com>
Date: Thu, 31 Jan 2019 11:24:08 -0800
Subject: [PATCH 24/24] Fix the test cases after refactoring

---
 .../scala/org/apache/livy/server/batch/BatchServletSpec.scala   | 2 +-
 .../livy/server/interactive/InteractiveSessionServletSpec.scala | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
index 505a61698..ed298000f 100644
--- a/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/batch/BatchServletSpec.scala
@@ -83,7 +83,7 @@ class BatchServletSpec extends BaseSessionServletSpec[BatchSession, BatchRecover
       .asInstanceOf[BatchSessionView]
 
     view.id shouldEqual id
-    Option(view.name) shouldEqual name
+    view.name shouldEqual name
     view.state shouldEqual state.toString
     view.appId shouldEqual Some(appId)
     view.appInfo shouldEqual appInfo
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
index 6c97debfe..0b061fa8d 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionServletSpec.scala
@@ -152,7 +152,7 @@ class InteractiveSessionServletSpec extends BaseInteractiveServletSpec {
     }
   }
 
-  Seq(Some("TEST-interactive-session"), None, null)
+  Seq(Some("TEST-interactive-session"), None)
     .foreach { case name =>
       it(s"should show session properties (name =$name") {
         testShowSessionProperties(name: Option[String])