/** Start the endpoint and receiver execution thread. */
defstart(): Unit = synchronized {
  if(isTrackerStarted) {
    throw newSparkException("ReceiverTracker already started")
  }

  if(!receiverInputStreams.isEmpty) {
    endpoint = ssc.env.rpcEnv.setupEndpoint(
      "ReceiverTracker", newReceiverTrackerEndpoint(ssc.env.rpcEnv))
    if(!skipReceiverLaunch) launchReceivers()
    logInfo("ReceiverTracker started")
    trackerState = Started
  }
}

/**
 * Get the receivers from the ReceiverInputDStreams, distributes them to the
 * worker nodes as a parallel collection, and runs them.
 */
private deflaunchReceivers(): Unit = {
  valreceivers = receiverInputStreams.map(nis => {
    valrcvr = nis.getReceiver()
    rcvr.setReceiverId(nis.id)
    rcvr
  })

  runDummySparkJob()

  logInfo("Starting "+ receivers.length + " receivers")
  endpoint.send(StartAllReceivers(receivers))
}

override defreceive: PartialFunction[Any, Unit] = {
  // Local messages
  caseStartAllReceivers(receivers) =>
    valscheduledLocations = schedulingPolicy.scheduleReceivers(receivers, getExecutors)
    for(receiver <- receivers) {
      valexecutors = scheduledLocations(receiver.streamId)
      updateReceiverScheduledExecutors(receiver.streamId, executors)
      receiverPreferredLocations(receiver.streamId) = receiver.preferredLocation
      startReceiver(receiver, executors)
    }

/**
 * Start a receiver along with its scheduled executors
 */
private defstartReceiver(
    receiver: Receiver[_],
    scheduledLocations: Seq[TaskLocation]): Unit = {
  defshouldStartReceiver: Boolean = {
    // It's okay to start when trackerState is Initialized or Started
    !(isTrackerStopping || isTrackerStopped)
  }

  valreceiverId = receiver.streamId
  if(!shouldStartReceiver) {
    onReceiverJobFinish(receiverId)
    return
  }

  valcheckpointDirOption = Option(ssc.checkpointDir)
  valserializableHadoopConf =
    newSerializableConfiguration(ssc.sparkContext.hadoopConfiguration)

  // Function to start the receiver on the worker node
  valstartReceiverFunc: Iterator[Receiver[_]] => Unit =
    (iterator: Iterator[Receiver[_]]) => {
      if(!iterator.hasNext) {
        throw newSparkException(
          "Could not start receiver as object not found.")
      }
      if(TaskContext.get().attemptNumber() == 0) {
        valreceiver = iterator.next()
        assert(iterator.hasNext == false)
        valsupervisor = newReceiverSupervisorImpl(
          receiver, SparkEnv.get, serializableHadoopConf.value, checkpointDirOption)
        supervisor.start()
        supervisor.awaitTermination()
      } else{
        // It's restarted by TaskScheduler, but we want to reschedule it again. So exit it.
      }
    }

  // Create the RDD using the scheduledLocations to run the receiver in a Spark job
  valreceiverRDD: RDD[Receiver[_]] =
    if(scheduledLocations.isEmpty) {
      ssc.sc.makeRDD(Seq(receiver), 1)
    } else{
      valpreferredLocations = scheduledLocations.map(_.toString).distinct
      ssc.sc.makeRDD(Seq(receiver -> preferredLocations))
    }
  receiverRDD.setName(s"Receiver$receiverId")
  ssc.sparkContext.setJobDescription(s"Streaming job running receiver$receiverId")
  ssc.sparkContext.setCallSite(Option(ssc.getStartSite()).getOrElse(Utils.getCallSite()))

  valfuture = ssc.sparkContext.submitJob[Receiver[_], Unit, Unit](
    receiverRDD, startReceiverFunc, Seq(0), (_, _) => Unit, ())
  // We will keep restarting the receiver job until ReceiverTracker is stopped
  future.onComplete {
    caseSuccess(_) =>
      if(!shouldStartReceiver) {
        onReceiverJobFinish(receiverId)
      } else{
        logInfo(s"Restarting Receiver$receiverId")
        self.send(RestartReceiver(receiver))
      }
    caseFailure(e) =>
      if(!shouldStartReceiver) {
        onReceiverJobFinish(receiverId)
      } else{
        logError("Receiver has been stopped. Try to restart it.", e)
        logInfo(s"Restarting Receiver$receiverId")
        self.send(RestartReceiver(receiver))
      }
  }(submitJobThreadPool)
  logInfo(s"Receiver${receiver.streamId} started")
}