Fixed DiskBlockManager to cleanup local directory if it's the driver

Author: sarutak

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -134,12 +134,6 @@ object SparkSubmit {
       }
     }
 
-    sysProps.put("spark.deploy.mode",
-      deployMode match {
-        case CLIENT => "client"
-        case _ => "cluster"
-      })
-
     // The following modes are not supported or applicable
     (clusterManager, deployMode) match {
       case (MESOS, CLUSTER) =>

core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala

@@ -166,8 +166,7 @@ private[spark] class DiskBlockManager(blockManager: BlockManager, conf: SparkCon
   /** Cleanup local dirs and stop shuffle sender. */
   private[spark] def stop() {
     // Only perform cleanup if an external service is not serving our shuffle files.
-    if (!blockManager.externalShuffleServiceEnabled ||
-      (blockManager.blockManagerId.isDriver && conf.get("spark.deploy.mode") == "client")) {
+    if (!blockManager.externalShuffleServiceEnabled || blockManager.blockManagerId.isDriver) {
       localDirs.foreach { localDir =>
         if (localDir.isDirectory() && localDir.exists()) {
           try {