Support asking driver status when it's retrying.

Author: tnachen

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala

@@ -446,6 +446,8 @@ private[spark] class MesosClusterScheduler(
         return StatusResponse(submissionId, true, Option("Driver is running"))
       } else if (finishedDrivers.contains(submissionId)) {
         return StatusResponse(submissionId, true, Option("Driver already finished"))
+      } else if (superviseRetryList.contains(submissionId)) {
+        return StatusResponse(submissionId, true, Option("Driver is retrying"))
       } else {
         return StatusResponse(submissionId, false, None)
       }

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/SuperviseRetryList.scala

@@ -46,6 +46,9 @@ private[mesos] class SuperviseRetryList(state: ClusterPersistenceEngine) {
     }
   }
 
+  def contains(submissionId: String) =
+    drivers.exists(d => d.submission.submissionId.equals(submissionId))
+
   def getNextRetry(currentTime: Date): (Option[DriverSubmission], Option[RetryState]) = {
     val retry = drivers.find(d => d.nextRetry.before(currentTime))
     if (retry.isDefined) {