Browse Source

HUE-2831 [livy] Don't report a connection error when killing a session

Erick Tryzelaar 10 years ago
parent
commit
7b0a4bcd32

+ 9 - 3
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/interactive/InteractiveSessionProcess.scala

@@ -21,14 +21,13 @@ package com.cloudera.hue.livy.server.interactive
 import java.lang.ProcessBuilder.Redirect
 import java.net.URL
 
-import com.cloudera.hue.livy.sessions.Error
+import com.cloudera.hue.livy.sessions.{Success, Dead, Error}
+import com.cloudera.hue.livy.spark.SparkSubmitProcessBuilder.{AbsolutePath, RelativePath}
 import com.cloudera.hue.livy.spark.{SparkProcess, SparkSubmitProcessBuilder}
-import com.cloudera.hue.livy.spark.SparkSubmitProcessBuilder.{RelativePath, AbsolutePath}
 import com.cloudera.hue.livy.{LivyConf, Logging, Utils}
 
 import scala.annotation.tailrec
 import scala.concurrent.Future
-import scala.io.Source
 
 object InteractiveSessionProcess extends Logging {
 
@@ -124,6 +123,13 @@ private class InteractiveSessionProcess(id: Int,
   Future {
     if (process.waitFor() != 0) {
       _state = Error()
+    } else {
+      // Set the state to done if the session shut down before contacting us.
+      _state match {
+        case (Dead() | Error() | Success()) =>
+        case _ =>
+          _state = Success()
+      }
     }
   }
 

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

@@ -18,7 +18,7 @@
 
 package com.cloudera.hue.livy.server.interactive
 
-import java.net.URL
+import java.net.{ConnectException, URL}
 import java.util.concurrent.TimeUnit
 
 import com.cloudera.hue.livy._
@@ -32,6 +32,7 @@ import org.json4s.{DefaultFormats, Formats, JValue}
 import scala.annotation.tailrec
 import scala.concurrent.duration.Duration
 import scala.concurrent.{Future, _}
+import scala.util
 
 abstract class InteractiveWebSession(val id: Int, createInteractiveRequest: CreateInteractiveRequest) extends InteractiveSession with Logging {
 
@@ -123,12 +124,18 @@ abstract class InteractiveWebSession(val id: Int, createInteractiveRequest: Crea
         case Idle() =>
           _state = Busy()
 
-          Http(svc.DELETE OK as.String).map { case rep =>
-            synchronized {
-              _state = Dead()
-            }
+          Http(svc.DELETE OK as.String).either() match {
+            case (Right(_) | Left(_: ConnectException)) =>
+              // Make sure to eat any connection errors because the repl shut down before it sent
+              // out an OK.
+              synchronized {
+                _state = Dead()
+              }
 
-            Unit
+              Future.successful(())
+
+            case Left(t: Throwable) =>
+              Future.failed(t)
           }
         case NotStarted() =>
           Future {