Stop SparkContext after stopping scheduler, not before.

Author: JoshRosen

streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala

@@ -492,10 +492,6 @@ class StreamingContext private[streaming] (
    *                       received data to be completed
    */
   def stop(stopSparkContext: Boolean, stopGracefully: Boolean): Unit = synchronized {
-    // Even if the streaming context has not been started, we still need to stop the SparkContext.
-    // Even if we have already stopped, we still need to attempt to stop the SparkContext because
-    // a user might stop(stopSparkContext = false) and then call stop(stopSparkContext = true).
-    if (stopSparkContext) sc.stop()
     if (state == Stopped) {
       logWarning("StreamingContext has already been stopped")
     } else {
@@ -507,6 +503,10 @@ class StreamingContext private[streaming] (
         waiter.notifyStop()
       }
     }
+    // Even if the streaming context has not been started, we still need to stop the SparkContext.
+    // Even if we have already stopped, we still need to attempt to stop the SparkContext because
+    // a user might stop(stopSparkContext = false) and then call stop(stopSparkContext = true).
+    if (stopSparkContext) sc.stop()
     // The state should always be Stopped after calling `stop()`, even if we haven't started yet:
     state = Stopped
   }