[STREAMING][TEST] Fix flaky streaming.FailureSuite

Under some corner cases, the test suite failed to shutdown the SparkContext causing cascaded failures. This fix does two things
- Makes sure no SparkContext is active after every test
- Makes sure StreamingContext is always shutdown (prevents leaking of StreamingContexts as well, just in case)

Author: Tathagata Das <tathagata.das1565@gmail.com>

Closes #11166 from tdas/fix-failuresuite.

2 files changed, 6 insertions, 2 deletions

diff --git a/streaming/src/test/scala/org/apache/spark/streaming/FailureSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/FailureSuite.scala
index 6a0b0a1d47..31e159e968 100644
--- a/streaming/src/test/scala/org/apache/spark/streaming/FailureSuite.scala
+++ b/streaming/src/test/scala/org/apache/spark/streaming/FailureSuite.scala
@@ -21,7 +21,7 @@ import java.io.File
 
 import org.scalatest.BeforeAndAfter
 
-import org.apache.spark.{Logging, SparkFunSuite}
+import org.apache.spark._
 import org.apache.spark.util.Utils
 
 /**
@@ -43,6 +43,9 @@ class FailureSuite extends SparkFunSuite with BeforeAndAfter with Logging {
       Utils.deleteRecursively(directory)
     }
     StreamingContext.getActive().foreach { _.stop() }
+
+    // Stop SparkContext if active
+    SparkContext.getOrCreate(new SparkConf().setMaster("local").setAppName("bla")).stop()
   }
 
   test("multiple failures with map") {
diff --git a/streaming/src/test/scala/org/apache/spark/streaming/MasterFailureTest.scala b/streaming/src/test/scala/org/apache/spark/streaming/MasterFailureTest.scala
index a02d49eced..faa9c4f0cb 100644
--- a/streaming/src/test/scala/org/apache/spark/streaming/MasterFailureTest.scala
+++ b/streaming/src/test/scala/org/apache/spark/streaming/MasterFailureTest.scala
@@ -242,6 +242,8 @@ object MasterFailureTest extends Logging {
         }
       } catch {
         case e: Exception => logError("Error running streaming context", e)
+      } finally {
+        ssc.stop()
       }
       if (killingThread.isAlive) {
         killingThread.interrupt()
@@ -250,7 +252,6 @@ object MasterFailureTest extends Logging {
         // to null after the next test creates the new SparkContext and fail the test.
         killingThread.join()
       }
-      ssc.stop()
 
       logInfo("Has been killed = " + killed)
       logInfo("Is last output generated = " + isLastOutputGenerated)