[CARMEL-6086] DAGScheduler exit because of jobId not exist (#1008)

Author: allenma

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

@@ -722,6 +722,7 @@ private[spark] class DAGScheduler(
    * @param job The job whose state to cleanup.
    */
   private def cleanupStateForJobAndIndependentStages(job: ActiveJob): Unit = {
+    logInfo(s"Clean up job:${job.jobId} state")
     val registeredStages = jobIdToStageIds.get(job.jobId)
     if (registeredStages.isEmpty || registeredStages.get.isEmpty) {
       logError("No stages registered for job " + job.jobId)
@@ -1452,19 +1453,21 @@ private[spark] class DAGScheduler(
             eventProcessLoop.post(SubmitMissingTask(
               stage,
               jobId,
+              properties,
               partitionsToCompute,
               taskIdToLocations,
               taskBinary,
               partitions))
           } else {
             handleSubmitMissingTask(
               SubmitMissingTask(
-              stage,
-              jobId,
-              partitionsToCompute,
-              taskIdToLocations,
-              taskBinary,
-              partitions))
+                stage,
+                jobId,
+                properties,
+                partitionsToCompute,
+                taskIdToLocations,
+                taskBinary,
+                partitions))
           }
         }
       }
@@ -1498,7 +1501,7 @@ private[spark] class DAGScheduler(
     }
     // Use the scheduling pool, job group, description, etc. from an ActiveJob associated
     // with this Stage
-    val properties = jobIdToActiveJob(missingTask.jobId).properties
+    val properties = missingTask.properties
 
     val tasks: Seq[Task[_]] = try {
       val serializedTaskMetrics = closureSerializer.serialize(stage.latestInfo.taskMetrics).array()

core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala

@@ -106,6 +106,7 @@ case class SpeculativeTaskSubmitted(task: Task[_]) extends DAGSchedulerEvent
 private[scheduler] case class SubmitMissingTask(
     stage: Stage,
     jobId: Int,
+    properties: Properties,
     partitionsToCompute: Seq[Int],
     taskIdToLocations: scala.collection.Map[Int, Seq[TaskLocation]],
     taskBinary: Broadcast[Array[Byte]],