Fix comments and code style

Author: zsxwing

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -1859,16 +1859,11 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
     new SimpleFutureAction(waiter, resultFunc)
   }
 
-  /**
-   * Submit a job for execution and return a FutureJob holding the result. Return a Future for
-   * monitoring the job success or failure event.
-   */
   private[spark] def submitAsyncJob[T, U, R](
       rdd: RDD[T],
       processPartition: (TaskContext, Iterator[T]) => U,
       resultHandler: (Int, U) => Unit,
-      resultFunc: => R): Future[Unit] =
-  {
+      resultFunc: => R): Future[Unit] = {
     assertNotStopped()
     val cleanF = clean(processPartition)
     val callSite = getCallSite

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

@@ -174,7 +174,7 @@ private[streaming] abstract class ReceiverSupervisor(
     }(futureExecutionContext)
   }
 
-  /** Reschedule this receiver and return a candidate executor list */
+  /** Return a list of candidate executors to run the receiver */
   def getAllowedLocations(): Seq[String] = Seq.empty
 
   /** Check if receiver has been marked for stopping */

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

@@ -31,8 +31,8 @@ private[streaming] case class ReceiverTrackingInfo(
 private[streaming] trait ReceiverScheduler {
 
   /**
-   * Return a candidate executor list to run the receiver. If the list is empty, the caller can run
-   * this receiver in arbitrary executor.
+   * Return a list of candidate executors to run the receiver. If the list is empty, the caller can
+   * run this receiver in arbitrary executor.
    */
   def scheduleReceiver(
       receiverId: Int,
@@ -69,7 +69,7 @@ private[streaming] class LoadBalanceReceiverSchedulerImpl extends ReceiverSchedu
           val scheduledLocations = receiverTrackingInfo.scheduledLocations.get
           // The probability that a scheduled receiver will run in an executor is
           // 1.0 / scheduledLocations.size
-          scheduledLocations.map(location => (location, 1.0 / scheduledLocations.size))
+          scheduledLocations.map(location => location -> 1.0 / scheduledLocations.size)
         case ReceiverState.ACTIVE => Seq(receiverTrackingInfo.runningLocation.get -> 1.0)
       }
     }.groupBy(_._1).mapValues(_.map(_._2).sum) // Sum weights for each executor

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

@@ -93,13 +93,17 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
 
   @volatile private var stopping = false
 
-  // Track receivers' status for scheduling
+  /**
+   * Track receivers' status for scheduling
+   */
   private val receiverTrackingInfos = new HashMap[Int, ReceiverTrackingInfo]
 
-  // Store all preferred locations for all receivers. We need this information to schedule receivers
+  /**
+   * Store all preferred locations for all receivers. We need this information to schedule receivers
+   */
   private val receiverPreferredLocations = new HashMap[Int, Option[String]]
 
-  // Use a separate lock to avoid dead-lock
+  /** Use a separate lock to avoid dead-lock */
   private val receiverTrackingInfosLock = new AnyRef
 
   /** Start the endpoint and receiver execution thread. */
@@ -405,8 +409,6 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
               }
             val future = ssc.sparkContext.submitAsyncJob[Receiver[_], Unit, Unit](
               receiverRDD, startReceiver, (_, _) => Unit, ())
-            // TODO Refactor JobWaiter to avoid creating a new Thread here. Otherwise, it's not
-            // scalable because we need a thread for each Receiver.
             future.onComplete {
               case Success(_) =>
                 if (stopping) {