Address minor code style comments

Author: zsxwing

streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceiverSupervisor.scala

@@ -99,7 +99,7 @@ private[streaming] abstract class ReceiverSupervisor(
   protected def onStop(message: String, error: Option[Throwable]) { }
 
   /** Called when receiver is started. Return true if the driver accepts us */
-  protected def onReceiverStart(): Boolean = true
+  protected def onReceiverStart(): Boolean
 
   /** Called when receiver is stopped */
   protected def onReceiverStop(message: String, error: Option[Throwable]) { }
@@ -129,7 +129,7 @@ private[streaming] abstract class ReceiverSupervisor(
         logInfo("Called receiver onStart")
       } else {
         // The driver refused us
-        stop("Registered unsuccessfully because the driver refused" + streamId, None)
+        stop("Registered unsuccessfully because Driver refused to start receiver " + streamId, None)
       }
     } catch {
       case NonFatal(t) =>

streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala

@@ -74,7 +74,7 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
 
   /** Enumeration to identify current state of the ReceiverTracker */
   object TrackerState extends Enumeration {
-    type CheckpointState = Value
+    type TrackerState = Value
     val Initialized, Started, Stopping, Stopped = Value
   }
   import TrackerState._
@@ -86,15 +86,6 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
   // This not being null means the tracker has been started and not stopped
   private var endpoint: RpcEndpointRef = null
 
-  /** Check if tracker has been marked for starting */
-  private def isTrackerStarted(): Boolean = trackerState == Started
-
-  /** Check if tracker has been marked for stopping */
-  private def isTrackerStopping(): Boolean = trackerState == Stopping
-
-  /** Check if tracker has been marked for stopped */
-  private def isTrackerStopped(): Boolean = trackerState == Stopped
-
   /** Start the endpoint and receiver execution thread. */
   def start(): Unit = synchronized {
     if (isTrackerStarted) {
@@ -412,4 +403,14 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
     }
 
   }
+
+  /** Check if tracker has been marked for starting */
+  private def isTrackerStarted(): Boolean = trackerState == Started
+
+  /** Check if tracker has been marked for stopping */
+  private def isTrackerStopping(): Boolean = trackerState == Stopping
+
+  /** Check if tracker has been marked for stopped */
+  private def isTrackerStopped(): Boolean = trackerState == Stopped
+
 }

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

@@ -346,6 +346,8 @@ class ReceiverSuite extends TestSuiteBase with Timeouts with Serializable {
     def reportError(message: String, throwable: Throwable) {
       errors += throwable
     }
+
+    override protected def onReceiverStart(): Boolean = true
   }
 
   /**

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

@@ -286,6 +286,8 @@ class StreamingContextSuite extends SparkFunSuite with BeforeAndAfter with Timeo
   }
 
   test("stop gracefully even if a receiver misses StopReceiver") {
+    // This is not a deterministic unit. But if this unit test is flaky, then there is definitely
+    // something wrong. See SPARK-5681
     val conf = new SparkConf().setMaster(master).setAppName(appName)
     sc = new SparkContext(conf)
     ssc = new StreamingContext(sc, Milliseconds(100))