[SPARK-2324] SparkContext should not exit directly when spark.local.dir is a list of multiple paths and one of them has error

Author: YanTangZhai

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

@@ -26,6 +26,8 @@ import org.apache.spark.executor.ExecutorExitCode
 import org.apache.spark.network.netty.{PathResolver, ShuffleSender}
 import org.apache.spark.util.Utils
 
+import scala.collection.mutable.ArrayBuffer
+
 /**
  * Creates and maintains the logical mapping between logical blocks and physical on-disk
  * locations. By default, one block is mapped to one file with a name given by its BlockId.
@@ -44,6 +46,10 @@ private[spark] class DiskBlockManager(shuffleManager: ShuffleBlockManager, rootD
    * directory, create multiple subdirectories that we will hash files into, in order to avoid
    * having really large inodes at the top level. */
   private val localDirs: Array[File] = createLocalDirs()
+  if (localDirs.isEmpty) {
+    logError("Failed to create any local dir")
+    System.exit(ExecutorExitCode.DISK_STORE_FAILED_TO_CREATE_DIR)
+  }
   private val subDirs = Array.fill(localDirs.length)(new Array[File](subDirsPerLocalDir))
   private var shuffleSender : ShuffleSender = null
 
@@ -115,8 +121,9 @@ private[spark] class DiskBlockManager(shuffleManager: ShuffleBlockManager, rootD
 
   private def createLocalDirs(): Array[File] = {
     logDebug(s"Creating local directories at root dirs '$rootDirs'")
+    val localDirsResult = ArrayBuffer[File]()
     val dateFormat = new SimpleDateFormat("yyyyMMddHHmmss")
-    rootDirs.split(",").map { rootDir =>
+    rootDirs.split(",").foreach { rootDir =>
       var foundLocalDir = false
       var localDir: File = null
       var localDirId: String = null
@@ -137,11 +144,12 @@ private[spark] class DiskBlockManager(shuffleManager: ShuffleBlockManager, rootD
       }
       if (!foundLocalDir) {
         logError(s"Failed $MAX_DIR_CREATION_ATTEMPTS attempts to create local dir in $rootDir")
-        System.exit(ExecutorExitCode.DISK_STORE_FAILED_TO_CREATE_DIR)
+      } else {
+        logInfo(s"Created local directory at $localDir")
+        localDirsResult += localDir
       }
-      logInfo(s"Created local directory at $localDir")
-      localDir
     }
+    localDirsResult.toArray
   }
 
   private def addShutdownHook() {