[CARMEL-6314] Make sure registerExecutor event must be handled in HeartBeatReceiver (#1097)

Author: allenma

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

@@ -175,13 +175,6 @@ private[spark] class HeartbeatReceiver(sc: SparkContext, clock: Clock)
     Option(self).map(_.ask[Boolean](ExecutorRegistered(executorId)))
   }
 
-  /**
-   * If the heartbeat receiver is not stopped, notify it of executor registrations.
-   */
-  override def onExecutorAdded(executorAdded: SparkListenerExecutorAdded): Unit = {
-    addExecutor(executorAdded.executorId)
-  }
-
   /**
    * Send ExecutorRemoved to the event loop to remove an executor. Only for test.
    *

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

@@ -262,6 +262,7 @@ class SparkContext(config: SparkConf) extends Logging {
   def master: String = _conf.get("spark.master")
   def deployMode: String = _conf.get(SUBMIT_DEPLOY_MODE)
   def appName: String = _conf.get("spark.app.name")
+  def heartbeatReceiverRef: RpcEndpointRef = _heartbeatReceiver
 
   private[spark] def isEventLogEnabled: Boolean = _conf.get(EVENT_LOG_ENABLED)
   private[spark] def eventLogDir: Option[URI] = _eventLogDir

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -29,7 +29,7 @@ import scala.util.Failure
 
 import org.apache.hadoop.security.UserGroupInformation
 
-import org.apache.spark.{ExecutorAllocationClient, SparkEnv, SparkException, TaskState}
+import org.apache.spark.{ExecutorAllocationClient, ExecutorRegistered, HeartbeatReceiver, SparkEnv, SparkException, TaskState}
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.deploy.security.HadoopDelegationTokenManager
 import org.apache.spark.executor.ExecutorLogUrlHandler
@@ -150,6 +150,8 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
   @volatile
   protected var lastOfferWarnLogTime = 0L
 
+  private val heartbeatReceiverRef = scheduler.sc.heartbeatReceiverRef
+
   class DriverEndpoint extends IsolatedRpcEndpoint with Logging {
 
     override val rpcEnv: RpcEnv = CoarseGrainedSchedulerBackend.this.rpcEnv
@@ -298,6 +300,8 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
               logDebug(s"Decremented number of pending executors ($numPendingExecutors left)")
             }
           }
+          // notify the heartbeatReceiver the executor is registered
+          heartbeatReceiverRef.ask[Boolean](ExecutorRegistered(executorId))
           listenerBus.post(
             SparkListenerExecutorAdded(System.currentTimeMillis(), executorId, data))
           // Note: some tests expect the reply to come after we put the executor in the map