Address more review comments

Author: JoshRosen

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

@@ -492,16 +492,13 @@ class StreamingContext private[streaming] (
    *                       received data to be completed
    */
   def stop(stopSparkContext: Boolean, stopGracefully: Boolean): Unit = synchronized {
-    if (state == Stopped) {
-      logWarning("StreamingContext has already been stopped")
-    } else {
-      if (state == Initialized) {
-        logWarning("StreamingContext has not been started yet")
-      } else {
+    state match {
+      case Initialized => logWarning("StreamingContext has not been started yet")
+      case Stopped => logWarning("StreamingContext has already been stopped")
+      case Started =>
         scheduler.stop(stopGracefully)
         logInfo("StreamingContext stopped successfully")
         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

streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala

@@ -139,18 +139,6 @@ class StreamingContextSuite extends FunSuite with BeforeAndAfter with Timeouts w
     ssc.stop()  // stop before start should not throw exception
   }
 
-  test("stop(stopContext=true) after stopSparkContext(stopContext=false)") {
-    ssc = new StreamingContext(master, appName, batchDuration)
-    addInputStream(ssc).register()
-    ssc.stop(stopSparkContext = false)
-    assert(ssc.sc.makeRDD(1 to 100).collect().size === 100)
-    ssc.stop(stopSparkContext = true)
-    // Check that the SparkContext is actually stopped:
-    intercept[Exception] {
-      ssc.sc.makeRDD(1 to 100).collect()
-    }
-  }
-
   test("start after stop") {
     // Regression test for SPARK-4301
     ssc = new StreamingContext(master, appName, batchDuration)
@@ -174,6 +162,18 @@ class StreamingContextSuite extends FunSuite with BeforeAndAfter with Timeouts w
     ssc.stop()
   }
 
+  test("stop(stopSparkContext=true) after stop(stopSparkContext=false)") {
+    ssc = new StreamingContext(master, appName, batchDuration)
+    addInputStream(ssc).register()
+    ssc.stop(stopSparkContext = false)
+    assert(ssc.sc.makeRDD(1 to 100).collect().size === 100)
+    ssc.stop(stopSparkContext = true)
+    // Check that the SparkContext is actually stopped:
+    intercept[Exception] {
+      ssc.sc.makeRDD(1 to 100).collect()
+    }
+  }
+
   test("stop gracefully") {
     val conf = new SparkConf().setMaster(master).setAppName(appName)
     conf.set("spark.cleaner.ttl", "3600")