Remove the ScheduleReceiver message because we can refuse it when receiving RegisterReceiver

Author: zsxwing

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

@@ -175,20 +175,12 @@ private[streaming] abstract class ReceiverSupervisor(
       stopReceiver("Restarting receiver with delay " + delay + "ms: " + message, error)
       logDebug("Sleeping for " + delay)
       Thread.sleep(delay)
-      val scheduledLocations = getAllowedLocations()
-      if (scheduledLocations.isEmpty || scheduledLocations.contains(host)) {
-        logInfo("Starting receiver again")
-        startReceiver()
-        logInfo("Receiver started again")
-      } else {
-        stop("Receiver is scheduled to another executor", None)
-      }
+      logInfo("Starting receiver again")
+      startReceiver()
+      logInfo("Receiver started again")
     }(futureExecutionContext)
   }
 
-  /** Return a list of candidate executors to run the receiver */
-  def getAllowedLocations(): Seq[String] = Seq.empty
-
   /** Check if receiver has been marked for starting */
   def isReceiverStarted(): Boolean = {
     logDebug("state = " + receiverState)

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

@@ -183,7 +183,4 @@ private[streaming] class ReceiverSupervisorImpl(
     receivedBlockHandler.cleanupOldBlocks(cleanupThreshTime.milliseconds)
   }
 
-  override def getAllowedLocations(): Seq[String] = {
-    trackerEndpoint.askWithRetry[Seq[String]](ScheduleReceiver(streamId))
-  }
 }

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

@@ -58,9 +58,6 @@ private[streaming] case class ReportError(streamId: Int, message: String, error:
 private[streaming] case class DeregisterReceiver(streamId: Int, msg: String, error: String)
   extends ReceiverTrackerMessage
 
-/** It's used to ask ReceiverTracker to return a candidate executor list to run the receiver */
-private[streaming] case class ScheduleReceiver(streamId: Int) extends ReceiverTrackerMessage
-
 /**
  * This class manages the execution of the receivers of ReceiverInputDStreams. Instance of
  * this class must be created after all input streams have been added and StreamingContext.start()
@@ -122,12 +119,12 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
   private def isTrackerStarted(): Boolean = trackerStateLock.synchronized {
     trackerState == Started
   }
- 
+
   /** Check if tracker has been marked for stopping */
   private def isTrackerStopping(): Boolean = trackerStateLock.synchronized {
     trackerState == Stopping
   }
- 
+
   /** Check if tracker has been marked for stopped */
   private def isTrackerStopped(): Boolean = trackerStateLock.synchronized {
     trackerState == Stopped
@@ -219,6 +216,9 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
     trackerStateLock.synchronized {
       if (isTrackerStopping || isTrackerStopped) {
         false
+      } else if (!ssc.sparkContext.isLocal && // We don't need to schedule it in the local mode
+        !scheduleReceiver(streamId).contains(host)) {
+        false
       } else {
         // When updating "receiverInfo", we should make sure "trackerState" won't be changed at the
         // same time. Therefore the following line should be in "trackerStateLock.synchronized".
@@ -305,8 +305,6 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
       case DeregisterReceiver(streamId, message, error) =>
         deregisterReceiver(streamId, message, error)
         context.reply(true)
-      case ScheduleReceiver(streamId) =>
-        context.reply(scheduleReceiver(streamId))
     }
   }
 

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

@@ -131,9 +131,7 @@ class ReceiverSuite extends TestSuiteBase with Timeouts with Serializable {
 
   test("restart receiver should consider the scheduled locations") {
     val receiver = new FakeReceiver
-    val executor = new FakeReceiverSupervisor(receiver) {
-      override def getAllowedLocations: Seq[String] = Seq("unknown-host")
-    }
+    val executor = new FakeReceiverSupervisor(receiver)
     executor.start()
     receiver.restart("force the receiver restart")
     eventually(timeout(30000 millis), interval(10 millis)) {