[SPARK-21660][YARN][Shuffle] Yarn ShuffleService failed to start when the chosen dir…

common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java

@@ -332,34 +332,64 @@ protected Path getRecoveryPath(String fileName) {
     return _recoveryPath;
   }
 
+  /**
+   * Check the chosen DB file available or not.
+   */
+  protected Boolean checkFileAvailable(File file) {
+      if (file.canWrite()){
+        return true;
+      }
+
+      return false;
+  }
+
   /**
    * Figure out the recovery path and handle moving the DB if YARN NM recovery gets enabled
    * when it previously was not. If YARN NM recovery is enabled it uses that path, otherwise
    * it will uses a YARN local dir.
    */
   protected File initRecoveryDb(String dbName) {
+    Boolean bolRecoveryPathAvailable = true;
+
     if (_recoveryPath != null) {
         File recoveryFile = new File(_recoveryPath.toUri().getPath(), dbName);
-        if (recoveryFile.exists()) {
+
+        bolRecoveryPathAvailable = checkFileAvailable(recoveryFile);
+        logger.info("Recovery path {} ldb available: {}.", _recoveryPath, bolRecoveryPathAvailable);
+        if (recoveryFile.exists() && bolRecoveryPathAvailable) {
           return recoveryFile;
         }
     }
+
+    // If recovery path unavailable, no use it any more.
+    if (!bolRecoveryPathAvailable) {
+      logger.warn("Recovery path {} unavailable: set it to null", _recoveryPath);
+      _recoveryPath = null;
+    }
+
     // db doesn't exist in recovery path go check local dirs for it
     String[] localDirs = _conf.getTrimmedStrings("yarn.nodemanager.local-dirs");
     for (String dir : localDirs) {
       File f = new File(new Path(dir).toUri().getPath(), dbName);
+      // 1. `_recoveryPath` not exists, `f` should be writable;
+      // 2. `_recoveryPath` exists, `newLoc` should be writable;
       if (f.exists()) {
         if (_recoveryPath == null) {
-          // If NM recovery is not enabled, we should specify the recovery path using NM local
-          // dirs, which is compatible with the old code.
-          _recoveryPath = new Path(dir);
-          return f;
+          if (checkFileAvailable(f)) {
+            // If NM recovery is not enabled, we should specify the recovery path using NM local
+            // dirs, which is compatible with the old code.
+            _recoveryPath = new Path(dir);
+            return f;
+          }
         } else {
           // If the recovery path is set then either NM recovery is enabled or another recovery
           // DB has been initialized. If NM recovery is enabled and had set the recovery path
           // make sure to move all DBs to the recovery path from the old NM local dirs.
           // If another DB was initialized first just make sure all the DBs are in the same
           // location.
+          if (!bolRecoveryPathAvailable) {
+              continue;
+          }
           Path newLoc = new Path(_recoveryPath, dbName);
           Path copyFrom = new Path(f.toURI());
           if (!newLoc.equals(copyFrom)) {
@@ -378,6 +408,18 @@ protected File initRecoveryDb(String dbName) {
         }
       }
     }
+
+    // Find a local_dir which is writable, to avoid creating ldb in a read-only disk.
+    if (_recoveryPath == null) {
+      for (String dir : localDirs) {
+        File f = new File(dir);
+        if (checkFileAvailable(f)) {
+          _recoveryPath = new Path(dir);
+          break;
+        }
+      }
+    }
+
     if (_recoveryPath == null) {
       _recoveryPath = new Path(localDirs[0]);
     }