Updated code as per the comments

cloudera · May 16, 2017 · 75d8511 · 75d8511
1 parent e94f507
commit 75d8511
Show file tree

Hide file tree

Showing 4 changed files with 8 additions and 8 deletions.
diff --git a/server/src/main/scala/com/cloudera/livy/server/interactive/InteractiveSession.scala b/server/src/main/scala/com/cloudera/livy/server/interactive/InteractiveSession.scala
@@ -380,9 +380,9 @@ class InteractiveSession(
   heartbeat()
 
   private val app = mockApp.orElse {
-      val driverProcess = client.flatMap { c => Option(c.getDriverProcess) }
+    val driverProcess = client.flatMap { c => Option(c.getDriverProcess) }
         .map(new LineBufferedProcess(_, livyConf.getInt(LivyConf.SPARK_LOGS_SIZE)))
-      Option(SparkApp.create(appTag, appId, driverProcess, livyConf, Some(this)))
+    driverProcess.map(_ => SparkApp.create(appTag, appId, driverProcess, livyConf, Some(this)))
   }
 
   if (client.isEmpty) {

diff --git a/server/src/main/scala/com/cloudera/livy/utils/LineBufferedStream.scala b/server/src/main/scala/com/cloudera/livy/utils/LineBufferedStream.scala
@@ -38,11 +38,11 @@ class LineBufferedStream(inputStream: InputStream, logSize: Int) extends Logging
   private val thread = new Thread {
     override def run() = {
       val lines = Source.fromInputStream(inputStream).getLines()
-       for (line <- lines) {
+      for (line <- lines) {
+        info(s"stdout: $line")
         _lock.lock()
-        try {
-          info(s"stdout: $line")
-          if(logSize > 0) _lines add line
+         try {
+          _lines add line
           _condition.signalAll()
         } finally {
           _lock.unlock()

diff --git a/server/src/main/scala/com/cloudera/livy/utils/SparkApp.scala b/server/src/main/scala/com/cloudera/livy/utils/SparkApp.scala
@@ -92,8 +92,7 @@ object SparkApp {
     if (livyConf.isRunningOnYarn()) {
       new SparkYarnApp(uniqueAppTag, appId, process, listener, livyConf)
     } else {
-      // process is None in recovery mode
-      // require(process.isDefined, "process must not be None when Livy master is not YARN.")
+      require(process.isDefined, "process must not be None when Livy master is not YARN.")
       new SparkProcApp(process.get, listener)
     }
   }

diff --git a/server/src/main/scala/com/cloudera/livy/utils/SparkProcApp.scala b/server/src/main/scala/com/cloudera/livy/utils/SparkProcApp.scala
@@ -42,6 +42,7 @@ class SparkProcApp (
 
   override def log(): IndexedSeq[String] =
     ("stdout: " +: process.inputLines) ++ ("\nstderr: " +: process.errorLines)
+
   private def changeState(newState: SparkApp.State.Value) = {
     if (state != newState) {
       listener.foreach(_.stateChanged(state, newState))