[Core] LiveExecutor MemoryLeak in AppStatusListener when ExecutorLost happen

override def executorLost(execId: String, host: String, reason: ExecutorLossReason): Unit = {
  // Re-enqueue any tasks that ran on the failed executor if this is a shuffle map stage,
  // and we are not using an external shuffle server which could serve the shuffle outputs.
  // The reason is the next stage wouldn't be able to fetch the data from this dead executor
  // so we would need to rerun these tasks on other executors.
  if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie) {
    for ((tid, info) <- taskInfos if info.executorId == execId) {
      val index = info.index
      // We may have a running task whose partition has been marked as successful,
      // this partition has another task completed in another stage attempt.
      // We treat it as a running task and will call handleFailedTask later.
      if (successful(index) && !info.running && !killedByOtherAttempt.contains(tid)) {
        successful(index) = false
        copiesRunning(index) -= 1
        tasksSuccessful -= 1
        addPendingTask(index)
        // Tell the DAGScheduler that this task was resubmitted so that it doesn't think our
        // stage finishes when a total of tasks.size tasks finish.
        sched.dagScheduler.taskEnded(
          tasks(index), Resubmitted, null, Seq.empty, Array.empty, info)
      }
    }
  }

/** Was the specified executor active for any currently live stages? */
private def isExecutorActiveForLiveStages(exec: LiveExecutor): Boolean = {
  liveStages.values.asScala.exists { stage =>
    stage.info.submissionTime.getOrElse(0L) < exec.removeTime.getTime
  }
} 

override def onStageCompleted(event: SparkListenerStageCompleted): Unit = {
  .....

  // remove any dead executors that were not running for any currently active stages
  deadExecutors.retain((execId, exec) => isExecutorActiveForLiveStages(exec))
}