Reorganize ReceiverTracker to use an event loop for lock free

1. Add fields into ReceiverTrackingInfo to contain all necessary information for ReceiverInfo. When sending ReceiverInfo to StreamingListener, create ReceiverInfo from ReceiverTrackingInfo directly.
2. Move methods that use ReceiverTrackingInfo into ReceiverTrackerEndpoint so that we can maintain all ReceiverTrackingInfos in an event loop.
3. Move ReceiverTrackingInfo to a separate file ReceiverTrackingInfo.scala.
4. Address some code style issues

Author: zsxwing

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

@@ -182,5 +182,4 @@ private[streaming] class ReceiverSupervisorImpl(
     logDebug(s"Cleaning up blocks older then $cleanupThreshTime")
     receivedBlockHandler.cleanupOldBlocks(cleanupThreshTime.milliseconds)
   }
-
 }

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

@@ -28,7 +28,6 @@ import org.apache.spark.rpc.RpcEndpointRef
 case class ReceiverInfo(
     streamId: Int,
     name: String,
-    private[streaming] val endpoint: RpcEndpointRef,
     active: Boolean,
     location: String,
     lastErrorMessage: String = "",

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

@@ -17,29 +17,8 @@
 
 package org.apache.spark.streaming.scheduler
 
+import scala.collection.Map
 import scala.collection.mutable
-import scala.util.Random
-
-import org.apache.spark.streaming.scheduler.ReceiverState._
-
-private[streaming] case class ReceiverTrackingInfo(
-    receiverId: Int,
-    state: ReceiverState,
-    scheduledLocations: Option[Seq[String]],
-    runningLocation: Option[String])
-
-private[streaming] trait ReceiverSchedulingPolicy {
-
-  /**
-   * 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,
-      preferredLocation: Option[String],
-      receiverTrackingInfoMap: Map[Int, ReceiverTrackingInfo],
-      executors: Seq[String]): Seq[String]
-}
 
 /**
  * A ReceiverScheduler trying to balance executors' load. Here is the approach to schedule executors
@@ -65,8 +44,12 @@ private[streaming] trait ReceiverSchedulingPolicy {
  * </ol>
  *
  */
-private[streaming] class LoadBalanceReceiverSchedulingPolicyImpl extends ReceiverSchedulingPolicy {
+private[streaming] class ReceiverSchedulingPolicy {
 
+  /**
+   * 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,
       preferredLocation: Option[String],
@@ -80,17 +63,14 @@ private[streaming] class LoadBalanceReceiverSchedulingPolicyImpl extends Receive
     val locations = mutable.Set[String]()
     locations ++= preferredLocation
 
-    val executorWeights = receiverTrackingInfoMap.filter { case (id, _) =>
-      // Ignore the receiver to be scheduled. It may be still running.
-      id != receiverId
-    }.values.flatMap { receiverTrackingInfo =>
+    val executorWeights = receiverTrackingInfoMap.values.flatMap { receiverTrackingInfo =>
       receiverTrackingInfo.state match {
         case ReceiverState.INACTIVE => Nil
         case ReceiverState.SCHEDULED =>
           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