Merge pull request #891 from xiajunluan/SPARK-864

[SPARK-864]DAGScheduler Exception if we delete Worker and StandaloneExecutorBackend then add Worker

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

@@ -531,9 +531,16 @@ class DAGScheduler(
         tasks += new ResultTask(stage.id, stage.rdd, job.func, partition, locs, id)
       }
     }
+
+    val properties = if (idToActiveJob.contains(stage.jobId)) {
+      idToActiveJob(stage.jobId).properties
+    } else {
+      //this stage will be assigned to "default" pool
+      null
+    }
+
     // must be run listener before possible NotSerializableException
     // should be "StageSubmitted" first and then "JobEnded"
-    val properties = idToActiveJob(stage.jobId).properties
     listenerBus.post(SparkListenerStageSubmitted(stage, tasks.size, properties))
 
     if (tasks.size > 0) {