Fixed SparkSubmit.scala and DiskBlockManager.scala in order to delete local directories of the driver of local-mode when external shuffle service is enabled

Author: sarutak

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

@@ -134,6 +134,12 @@ 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,7 +166,8 @@ 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) {
+    if (!blockManager.externalShuffleServiceEnabled ||
+      (blockManager.blockManagerId.isDriver && conf.get("spark.deploy.mode") == "client")) {
       localDirs.foreach { localDir =>
         if (localDir.isDirectory() && localDir.exists()) {
           try {