MAPREDUCE-7474. Improve Manifest committer resilience

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/ManifestCommitterConstants.java

@@ -132,7 +132,9 @@ public final class ManifestCommitterConstants {
    * Should dir cleanup do parallel deletion of task attempt dirs
    * before trying to delete the toplevel dirs.
    * For GCS this may deliver speedup, while on ABFS it may avoid
-   * timeouts in certain deployments.
+   * timeouts in certain deployments, something
+   * {@link #OPT_CLEANUP_PARALLEL_DELETE_BASE_FIRST}
+   * can alleviate.
    * Value: {@value}.
    */
   public static final String OPT_CLEANUP_PARALLEL_DELETE =
@@ -143,6 +145,20 @@ public final class ManifestCommitterConstants {
    */
   public static final boolean OPT_CLEANUP_PARALLEL_DELETE_DIRS_DEFAULT = true;
 
+  /**
+   * Should parallel cleanup try to delete teh base first?
+   * Best for azure as it skips the task attempt deletions unless
+   * the toplevel delete fails.
+   * Value: {@value}.
+   */
+  public static final String OPT_CLEANUP_PARALLEL_DELETE_BASE_FIRST =
+      OPT_PREFIX + "cleanup.parallel.delete.base.first";
+
+  /**
+   * Default value of option {@link #OPT_CLEANUP_PARALLEL_DELETE_BASE_FIRST}:  {@value}.
+   */
+  public static final boolean OPT_CLEANUP_PARALLEL_DELETE_BASE_FIRST_DEFAULT = true;
+
   /**
    * Threads to use for IO.
    */

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/ManifestCommitterStatisticNames.java

@@ -259,6 +259,9 @@ public final class ManifestCommitterStatisticNames {
   public static final String OP_STAGE_TASK_SCAN_DIRECTORY
       = "task_stage_scan_directory";
 
+  /** Delete a directory: {@value}. */
+  public static final String OP_DELETE_DIR = "op_delete_dir";
+
   private ManifestCommitterStatisticNames() {
   }
 }

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/impl/InternalConstants.java

@@ -73,6 +73,7 @@ private InternalConstants() {
       OP_CREATE_ONE_DIRECTORY,
       OP_DIRECTORY_SCAN,
       OP_DELETE,
+      OP_DELETE_DIR,
       OP_DELETE_FILE_UNDER_DESTINATION,
       OP_GET_FILE_STATUS,
       OP_IS_DIRECTORY,
@@ -127,4 +128,17 @@ private InternalConstants() {
   /** Schemas of filesystems we know to not work with this committer. */
   public static final Set<String> UNSUPPORTED_FS_SCHEMAS =
       ImmutableSet.of("s3a", "wasb");
+
+  /**
+   * How many attempts to save a manifest by save and rename
+   * before giving up: {@value}.
+   */
+  public static final int SAVE_RETRY_COUNT = 5;
+
+  /**
+   * Interval in milliseconds between save retries.
+   * Value {@value} milliseconds.
+   */
+  public static final int SAVE_SLEEP_INTERVAL = 500;
+
 }

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/impl/ManifestStoreOperations.java

@@ -97,6 +97,36 @@ public boolean isFile(Path path) throws IOException {
   public abstract boolean delete(Path path, boolean recursive)
       throws IOException;
 
+  /**
+   * Forward to {@code delete(Path, true)}
+   * unless overridden.
+   * <p>
+   * If it returns without an error: there is no file at
+   * the end of the path.
+   * @param path path
+   * @return outcome
+   * @throws IOException failure.
+   */
+  public boolean deleteFile(Path path)
+      throws IOException {
+    return delete(path, false);
+  }
+
+  /**
+   * Acquire the delete capacity then call {@code FileSystem#delete(Path, true)}
+   * or equivalent.
+   * <p>
+   * If it returns without an error: there is nothing at
+   * the end of the path.
+   * @param path path
+   * @return outcome
+   * @throws IOException failure.
+   */
+  public boolean rmdir(Path path)
+      throws IOException {
+    return delete(path, true);
+  }
+
   /**
    * Forward to {@link FileSystem#mkdirs(Path)}.
    * Usual "what does 'false' mean" ambiguity.

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/impl/ManifestStoreOperationsThroughFileSystem.java

@@ -108,6 +108,11 @@ public boolean delete(Path path, boolean recursive)
     return fileSystem.delete(path, recursive);
   }
 
+  @Override
+  public boolean rmdir(final Path path) throws IOException {
+    return fileSystem.delete(path, true);
+  }
+
   @Override
   public boolean mkdirs(Path path)
       throws IOException {

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/stages/AbortTaskStage.java

@@ -25,6 +25,7 @@
 
 import org.apache.hadoop.fs.Path;
 
+import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_DELETE_DIR;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_STAGE_TASK_ABORT_TASK;
 
 /**
@@ -55,7 +56,11 @@ protected Path executeStage(final Boolean suppressExceptions)
     final Path dir = getTaskAttemptDir();
     if (dir != null) {
       LOG.info("{}: Deleting task attempt directory {}", getName(), dir);
-      deleteDir(dir, suppressExceptions);
+      if (suppressExceptions) {
+        deleteDirSuppressingExceptions(dir, OP_DELETE_DIR);
+      } else {
+        deleteDir(dir, OP_DELETE_DIR);
+      }
     }
     return dir;
   }

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/committer/manifest/stages/AbstractJobOrTaskStage.java

@@ -56,11 +56,14 @@
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterConstants.MANIFEST_SUFFIX;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_COMMIT_FILE_RENAME;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_COMMIT_FILE_RENAME_RECOVERED;
+import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_DELETE_DIR;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_LOAD_MANIFEST;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_MSYNC;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_RENAME_FILE;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterStatisticNames.OP_SAVE_TASK_MANIFEST;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.impl.AuditingIntegration.enterStageWorker;
+import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.impl.InternalConstants.SAVE_RETRY_COUNT;
+import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.impl.InternalConstants.SAVE_SLEEP_INTERVAL;
 
 /**
  * A Stage in Task/Job Commit.
@@ -445,9 +448,29 @@ protected Boolean delete(
       final boolean recursive,
       final String statistic)
       throws IOException {
-    return trackDuration(getIOStatistics(), statistic, () -> {
-      return operations.delete(path, recursive);
-    });
+    if (recursive) {
+      return deleteDir(path, statistic);
+    } else {
+      return deleteFile(path, statistic);
+    }
+  }
+
+  /**
+   * Delete a file at a path.
+   * <p>
+   * If it returns without an error: there is nothing at
+   * the end of the path.
+   * @param path path
+   * @param statistic statistic to update
+   * @return outcome.
+   * @throws IOException IO Failure.
+   */
+  protected boolean deleteFile(
+      final Path path,
+      final String statistic)
+      throws IOException {
+    return trackDuration(getIOStatistics(), statistic, () ->
+        operations.deleteFile(path));
   }
 
   /**
@@ -582,19 +605,46 @@ protected final Path directoryMustExist(
    * Save a task manifest or summary. This will be done by
    * writing to a temp path and then renaming.
    * If the destination path exists: Delete it.
+   * This will retry so that a rename failure from abfs load or IO errors
+   * will not fail the task.
    * @param manifestData the manifest/success file
    * @param tempPath temp path for the initial save
    * @param finalPath final path for rename.
-   * @throws IOException failure to load/parse
+   * @throws IOException failure to rename after retries.
    */
   @SuppressWarnings("unchecked")
   protected final <T extends AbstractManifestData> void save(T manifestData,
       final Path tempPath,
       final Path finalPath) throws IOException {
-    LOG.trace("{}: save('{}, {}, {}')", getName(), manifestData, tempPath, finalPath);
-    trackDurationOfInvocation(getIOStatistics(), OP_SAVE_TASK_MANIFEST, () ->
-        operations.save(manifestData, tempPath, true));
-    renameFile(tempPath, finalPath);
+    boolean success = false;
+    int failures = 0;
+    while (!success) {
+      try {
+        LOG.trace("{}: attempt {} save('{}, {}, {}')",
+            getName(), failures, manifestData, tempPath, finalPath);
+
+        trackDurationOfInvocation(getIOStatistics(), OP_SAVE_TASK_MANIFEST, () ->
+            operations.save(manifestData, tempPath, true));
+        renameFile(tempPath, finalPath);
+        // success flag is only set after the rename.
+        success = true;
+      } catch (IOException e) {
+        LOG.warn("Failed to save and commit file {} renamed to {}",
+            tempPath, finalPath, e);
+        failures++;
+        if (failures >= SAVE_RETRY_COUNT) {
+          // too many failures: escalate.
+          throw e;
+        }
+        // else, sleep
+        try {
+          Thread.sleep(SAVE_SLEEP_INTERVAL);
+        } catch (InterruptedException ie) {
+          Thread.currentThread().interrupt();
+        }
+      }
+    }
+
   }
 
   /**
@@ -696,12 +746,15 @@ protected boolean storeSupportsResilientCommit() {
    */
   private void maybeDeleteDest(final boolean deleteDest, final Path dest) throws IOException {
 
-    if (deleteDest && getFileStatusOrNull(dest) != null) {
-
-      boolean deleted = delete(dest, true);
-      // log the outcome in case of emergency diagnostics traces
-      // being needed.
-      LOG.debug("{}: delete('{}') returned {}'", getName(), dest, deleted);
+    if (deleteDest) {
+      final FileStatus st = getFileStatusOrNull(dest);
+      if (st != null) {
+        if (st.isDirectory()) {
+          deleteDir(dest, OP_DELETE_DIR);
+        } else {
+          deleteFile(dest, OP_DELETE);
+        }
+      }
     }
   }
 
@@ -915,26 +968,36 @@ protected final TaskPool.Submitter getIOProcessors(int size) {
   }
 
   /**
-   * Delete a directory, possibly suppressing exceptions.
+   * Delete a directory.
    * @param dir directory.
-   * @param suppressExceptions should exceptions be suppressed?
+   * @param statistic statistic to use
+   * @return true if the path is no longer present.
    * @throws IOException exceptions raised in delete if not suppressed.
-   * @return any exception caught and suppressed
    */
-  protected IOException deleteDir(
+  protected boolean deleteDir(
+      final Path dir,
+      final String statistic)
+      throws IOException {
+    return trackDuration(getIOStatistics(), statistic, () ->
+        operations.rmdir(dir));
+  }
+
+  /**
+   * Delete a directory, suprressing exceptions.
+   * @param dir directory.
+   * @param statistic statistic to use
+   * @return any exception caught.
+   */
+  protected IOException deleteDirSuppressingExceptions(
       final Path dir,
-      final Boolean suppressExceptions)
+      final String statistic)
       throws IOException {
     try {
-      delete(dir, true);
+      deleteDir(dir, statistic);
       return null;
     } catch (IOException ex) {
       LOG.info("Error deleting {}: {}", dir, ex.toString());
-      if (!suppressExceptions) {
-        throw ex;
-      } else {
-        return ex;
-      }
+      return ex;
     }
   }