Fix tests.

Author: tnachen

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

@@ -60,10 +60,9 @@ private[mesos] class DriverQueue(state: ClusterPersistenceEngine, capacity: Int)
     val removed = queue.dequeueFirst(d => d.submissionId.equals(submissionId))
     if (removed.isDefined) {
       state.expunge(removed.get.submissionId)
-      true
     }
 
-    false
+    removed.isDefined
   }
 
   def peek(): Option[DriverSubmission] = {

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

@@ -165,7 +165,7 @@ private[spark] class MesosClusterScheduler(
     }
   }
 
-  private def recoverState {
+  def recoverState {
     stateLock.synchronized {
       queue = new DriverQueue(engineFactory.createEngine("driverQueue"), queuedCapacity)
 

core/src/test/scala/org/apache/spark/scheduler/mesos/MesosClusterSchedulerSuite.scala

@@ -37,6 +37,7 @@ class MesosClusterSchedulerSuite extends FunSuite with LocalSparkContext with Mo
     conf.setMaster("mesos://localhost:5050")
     conf.setAppName("spark mesos")
     val scheduler = new MesosClusterScheduler(new BlackHolePersistenceEngineFactory, conf)
+    scheduler.recoverState
     val response =
       scheduler.submitDriver(
         new MesosDriverDescription(new DriverDescription("jar", 1000, 1, true, createCommand),
@@ -59,6 +60,7 @@ class MesosClusterSchedulerSuite extends FunSuite with LocalSparkContext with Mo
     conf.setMaster("mesos://localhost:5050")
     conf.setAppName("spark mesos")
     val scheduler = new MesosClusterScheduler(new BlackHolePersistenceEngineFactory, conf)
+    scheduler.recoverState
     val response =
       scheduler.submitDriver(new MesosDriverDescription(
         new DriverDescription("jar", 1000, 1, true, createCommand),