From 58dba70ac26fb77c1668a259a347603947787afe Mon Sep 17 00:00:00 2001
From: Joshi <rekhajoshm@gmail.com>
Date: Wed, 24 Jun 2015 20:30:57 -0700
Subject: [PATCH] SPARK-2645: Fix for SparkContext stop behavior

---
 .../scala/org/apache/spark/SparkEnv.scala     |  2 +-
 .../org/apache/spark/SparkContextSuite.scala  | 21 +++++++++++++++++++
 2 files changed, 22 insertions(+), 1 deletion(-)

diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala b/core/src/main/scala/org/apache/spark/SparkEnv.scala
index edfe570897d1f..a4b511b4f0ea0 100644
--- a/core/src/main/scala/org/apache/spark/SparkEnv.scala
+++ b/core/src/main/scala/org/apache/spark/SparkEnv.scala
@@ -93,7 +93,7 @@ class SparkEnv (
 
   private[spark] def stop() {
 
-    if(!isStopped) {
+    if (!isStopped) {
       isStopped = true
       try {
         pythonWorkers.foreach { case (key, worker) => worker.stop()}
diff --git a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
index 6838b35ab4cc8..765356c715a60 100644
--- a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
@@ -30,6 +30,7 @@ import org.apache.spark.util.Utils
 
 import scala.concurrent.Await
 import scala.concurrent.duration.Duration
+import scala.util.control.NonFatal
 
 class SparkContextSuite extends SparkFunSuite with LocalSparkContext {
 
@@ -272,4 +273,24 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext {
       sc.stop()
     }
   }
+
+  test("calling multiple sc.stop() must not throw uncaught exception(50) from sparkenv") {
+    var threwNoOrOnlyExceptedException = true
+    try {
+      sc = new SparkContext(new SparkConf().setAppName("test").setMaster("local"))
+      val cnt = sc.parallelize(1 to 4).count()
+      sc.cancelAllJobs()
+      sc.stop()
+      // call stop second time
+      sc.stop()
+    } catch {
+      case e: ServerStateException =>
+        // assert(!e.getMessage.contains("Server is already stopped"))
+        threwNoOrOnlyExceptedException = false
+      case NonFatal(e) =>
+        threwNoOrOnlyExceptedException = true
+    } finally {
+      assert(threwNoOrOnlyExceptedException == true)
+    }
+  }
 }