diff --git a/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala
index 66e1e645007a..b630d333792d 100644
--- a/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcher.scala
@@ -59,7 +59,7 @@ private[mesos] class MesosClusterDispatcher(
     case _ => throw new IllegalArgumentException("Unsupported recovery mode: " + recoveryMode)
   }
 
-  private val scheduler = new MesosClusterScheduler(engineFactory, conf)
+  private val scheduler = new MesosClusterScheduler(engineFactory, conf, args.driverFailOver)
 
   private val server = new MesosRestServer(args.host, args.port, conf, scheduler)
   private val webUi = new MesosClusterUI(
diff --git a/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcherArguments.scala b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcherArguments.scala
index 5accaf78d0a5..5d9c41dc331c 100644
--- a/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcherArguments.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/mesos/MesosClusterDispatcherArguments.scala
@@ -29,6 +29,7 @@ private[mesos] class MesosClusterDispatcherArguments(args: Array[String], conf:
   var masterUrl: String = _
   var zookeeperUrl: Option[String] = None
   var propertiesFile: String = _
+  var driverFailOver = true // keep the driver registered with Mesos after dispatcher is stopped
 
   parse(args.toList)
 
@@ -70,6 +71,10 @@ private[mesos] class MesosClusterDispatcherArguments(args: Array[String], conf:
       propertiesFile = value
       parse(tail)
 
+    case ("--disable-failover") :: tail =>
+      driverFailOver = false
+      parse(tail)
+
     case ("--help") :: tail =>
       printUsageAndExit(0)
 
@@ -97,6 +102,7 @@ private[mesos] class MesosClusterDispatcherArguments(args: Array[String], conf:
         "  --webui-port WEBUI_PORT WebUI Port to listen on (default: 8081)\n" +
         "  --name NAME             Framework name to show in Mesos UI\n" +
         "  -m --master MASTER      URI for connecting to Mesos master\n" +
+        "  --disable-failover      De-register the framework when dispatcher is stopped\n" +
         "  -z --zk ZOOKEEPER       Comma delimited URLs for connecting to \n" +
         "                          Zookeeper for persistence\n" +
         "  --properties-file FILE  Path to a custom Spark properties file.\n" +
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
index 05fda0fded7f..1b863913f3ff 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
@@ -114,7 +114,8 @@ private[spark] class MesosDriverState(
  */
 private[spark] class MesosClusterScheduler(
     engineFactory: MesosClusterPersistenceEngineFactory,
-    conf: SparkConf)
+    conf: SparkConf,
+    driverFailOver: Boolean = true)
   extends Scheduler with MesosSchedulerUtils {
   var frameworkUrl: String = _
   private val metricsSystem =
@@ -306,8 +307,8 @@ private[spark] class MesosClusterScheduler(
       appName,
       conf,
       Some(frameworkUrl),
-      Some(true),
-      Some(Integer.MAX_VALUE),
+      Some(driverFailOver), // with checkpoint data if failOver is true
+      Some(if (driverFailOver) Double.MaxValue else 0.0), // timeout, 0.0 means no recovery
       fwId)
 
     startScheduler(driver)