MAPREDUCE-7474. review feedback and test improvements

Simulating more failure conditions.
Still more to explore there, in particular "what if delete of rename target fails"

Change-Id: Idb84f9c17a195702e6a2345b095f41e72865dd5b

Author: steveloughran

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

@@ -146,7 +146,7 @@ public final class ManifestCommitterConstants {
   public static final boolean OPT_CLEANUP_PARALLEL_DELETE_DIRS_DEFAULT = true;
 
   /**
-   * Should parallel cleanup try to delete teh base first?
+   * Should parallel cleanup try to delete the base first?
    * Best for azure as it skips the task attempt deletions unless
    * the toplevel delete fails.
    * Value: {@value}.
@@ -157,7 +157,7 @@ public final class ManifestCommitterConstants {
   /**
    * Default value of option {@link #OPT_CLEANUP_PARALLEL_DELETE_BASE_FIRST}:  {@value}.
    */
-  public static final boolean OPT_CLEANUP_PARALLEL_DELETE_BASE_FIRST_DEFAULT = true;
+  public static final boolean OPT_CLEANUP_PARALLEL_DELETE_BASE_FIRST_DEFAULT = false;
 
   /**
    * 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/impl/ManifestStoreOperations.java

@@ -113,16 +113,15 @@ public boolean deleteFile(Path path)
   }
 
   /**
-   * Acquire the delete capacity then call {@code FileSystem#delete(Path, true)}
-   * or equivalent.
+   * 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)
+  public boolean deleteRecursive(Path path)
       throws IOException {
     return delete(path, true);
   }

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

@@ -109,7 +109,7 @@ public boolean delete(Path path, boolean recursive)
   }
 
   @Override
-  public boolean rmdir(final Path path) throws IOException {
+  public boolean deleteRecursive(final Path path) throws IOException {
     return fileSystem.delete(path, true);
   }
 

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

@@ -57,9 +57,9 @@ protected Path executeStage(final Boolean suppressExceptions)
     if (dir != null) {
       LOG.info("{}: Deleting task attempt directory {}", getName(), dir);
       if (suppressExceptions) {
-        deleteDirSuppressingExceptions(dir, OP_DELETE_DIR);
+        deleteRecursiveSuppressingExceptions(dir, OP_DELETE_DIR);
       } else {
-        deleteDir(dir, OP_DELETE_DIR);
+        deleteRecursive(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

@@ -451,7 +451,7 @@ protected Boolean delete(
       final String statistic)
       throws IOException {
     if (recursive) {
-      return deleteDir(path, statistic);
+      return deleteRecursive(path, statistic);
     } else {
       return deleteFile(path, statistic);
     }
@@ -630,8 +630,15 @@ protected final <T extends AbstractManifestData> void save(T manifestData,
         LOG.info("{}: save manifest to {} then rename as {}'); retry count={}",
             getName(), tempPath, finalPath, retryCount);
 
+        // save the temp file, overwriting any which remains from an earlier attempt
         trackDurationOfInvocation(getIOStatistics(), OP_SAVE_TASK_MANIFEST, () ->
             operations.save(manifestData, tempPath, true));
+
+        // delete the destination in case it exists either from a failed previous
+        // attempt or from a concurrent task commit.
+        deleteFile(finalPath, OP_DELETE);
+
+        // rename temp to final
         renameFile(tempPath, finalPath);
         // success flag is only set after the rename.
         success = true;
@@ -650,6 +657,7 @@ protected final <T extends AbstractManifestData> void save(T manifestData,
           LOG.debug("Failure in retry policy", ex);
           retryAction = RetryPolicy.RetryAction.FAIL;
         }
+        LOG.debug("{}: Retry action: {}", getName(), retryAction.action);
         if (retryAction.action == RetryPolicy.RetryAction.RetryDecision.FAIL) {
           // too many failures: escalate.
           throw e;
@@ -679,16 +687,17 @@ public String getEtag(FileStatus status) {
   }
 
   /**
-   * Rename a file from source to dest; if the underlying FS API call
-   * returned false that's escalated to an IOE.
+   * Rename a file from source to dest.
+   * <p>
+   * The destination is always deleted through a call to
+   * {@link #maybeDeleteDest(boolean, Path)}.
    * @param source source file.
    * @param dest dest file
    * @throws IOException failure
    * @throws PathIOException if the rename() call returned false.
    */
   protected final void renameFile(final Path source, final Path dest)
       throws IOException {
-    maybeDeleteDest(true, dest);
     executeRenamingOperation("renameFile", source, dest,
         OP_RENAME_FILE, () ->
             operations.renameFile(source, dest));
@@ -770,7 +779,7 @@ private void maybeDeleteDest(final boolean deleteDest, final Path dest) throws I
       final FileStatus st = getFileStatusOrNull(dest);
       if (st != null) {
         if (st.isDirectory()) {
-          deleteDir(dest, OP_DELETE_DIR);
+          deleteRecursive(dest, OP_DELETE_DIR);
         } else {
           deleteFile(dest, OP_DELETE);
         }
@@ -988,32 +997,32 @@ protected final TaskPool.Submitter getIOProcessors(int size) {
   }
 
   /**
-   * Delete a directory.
+   * Delete a directory (or a file)
    * @param dir directory.
    * @param statistic statistic to use
    * @return true if the path is no longer present.
    * @throws IOException exceptions raised in delete if not suppressed.
    */
-  protected boolean deleteDir(
+  protected boolean deleteRecursive(
       final Path dir,
       final String statistic)
       throws IOException {
     return trackDuration(getIOStatistics(), statistic, () ->
-        operations.rmdir(dir));
+        operations.deleteRecursive(dir));
   }
 
   /**
-   * Delete a directory, suprressing exceptions.
+   * Delete a directory or file, catching exceptions.
    * @param dir directory.
    * @param statistic statistic to use
    * @return any exception caught.
    */
-  protected IOException deleteDirSuppressingExceptions(
+  protected IOException deleteRecursiveSuppressingExceptions(
       final Path dir,
       final String statistic)
       throws IOException {
     try {
-      deleteDir(dir, statistic);
+      deleteRecursive(dir, statistic);
       return null;
     } catch (IOException ex) {
       LOG.info("Error deleting {}: {}", dir, ex.toString());

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

@@ -174,9 +174,9 @@ protected Result executeStage(
             "Initial delete of %s", baseDir)) {
           exception = deleteOneDir(baseDir);
           if (exception == null) {
-            // success: record this as the outcome, which
-            // will skip the parallel delete.
+            // success: record this as the outcome,
             outcome = Outcome.DELETED;
+            // and will skip the parallel delete
             baseDirDeleted = true;
           } else {
             // failure: log and continue
@@ -276,7 +276,7 @@ private void rmTaskAttemptDir(FileStatus status) throws IOException {
   }
 
   /**
-   * Delete a directory.
+   * Delete a directory suppressing exceptions.
    * The {@link #deleteFailureCount} counter.
    * is incremented on every failure.
    * @param dir directory
@@ -288,7 +288,7 @@ private IOException deleteOneDir(final Path dir)
 
     deleteDirCount.incrementAndGet();
     return noteAnyDeleteFailure(
-        deleteDirSuppressingExceptions(dir, OP_DELETE_DIR));
+        deleteRecursiveSuppressingExceptions(dir, OP_DELETE_DIR));
   }
 
   /**

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

@@ -111,5 +111,9 @@ public TaskManifest getTaskManifest() {
       return taskManifest;
     }
 
+    @Override
+    public String toString() {
+      return "Result{path=" + path + '}';
+    }
   }
 }

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/site/markdown/manifest_committer.md

@@ -190,7 +190,7 @@ Here are the main configuration options of the committer.
 | `mapreduce.manifest.committer.io.threads` | Thread count for parallel operations | `64` |
 | `mapreduce.manifest.committer.summary.report.directory` | directory to save reports. | `""` |
 | `mapreduce.manifest.committer.cleanup.parallel.delete` | Delete temporary directories in parallel | `true` |
-| `mapreduce.manifest.committer.cleanup.parallel.delete.base.first` | Attempt to delete the base directory before parallel task attempts | `true` |
+| `mapreduce.manifest.committer.cleanup.parallel.delete.base.first` | Attempt to delete the base directory before parallel task attempts | `false` |
 | `mapreduce.fileoutputcommitter.cleanup.skipped` | Skip cleanup of `_temporary` directory| `false` |
 | `mapreduce.fileoutputcommitter.cleanup-failures.ignored` | Ignore errors during cleanup | `false` |
 | `mapreduce.fileoutputcommitter.marksuccessfuljobs` | Create a `_SUCCESS` marker file on successful completion. (and delete any existing one in job setup) | `true` |
@@ -423,7 +423,7 @@ may surface in cloud storage.
 | `mapreduce.fileoutputcommitter.cleanup.skipped` | Skip cleanup of `_temporary` directory| `false` |
 | `mapreduce.fileoutputcommitter.cleanup-failures.ignored` | Ignore errors during cleanup | `false` |
 | `mapreduce.manifest.committer.cleanup.parallel.delete` | Delete task attempt directories in parallel | `true` |
-| `mapreduce.manifest.committer.cleanup.parallel.delete.base.first` | Attempt to delete the base directory before parallel task attempts | `true` |
+| `mapreduce.manifest.committer.cleanup.parallel.delete.base.first` | Attempt to delete the base directory before parallel task attempts | `false` |
 
 The algorithm is:
 
@@ -444,7 +444,15 @@ if caught-exception and not "mapreduce.fileoutputcommitter.cleanup-failures.igno
 It's a bit complicated, but the goal is to perform a fast/scalable delete and
 throw a meaningful exception if that didn't work.
 
-For ABFS the default settings should normally be left alone.
+For ABFS set `mapreduce.manifest.committer.cleanup.parallel.delete.base.first` to `true`
+which should normally result in less network IO and a faster cleanup.
+
+```
+spark.hadoop.mapreduce.manifest.committer.cleanup.parallel.delete.base.first true
+```
+
+
+
 
 For GCS, setting `mapreduce.manifest.committer.cleanup.parallel.delete.base.first`
 to `false` may speed up cleanup.
@@ -490,9 +498,15 @@ The core set of Azure-optimized options becomes
 </property>
 
 <property>
-  <name>spark.hadoop.fs.azure.io.rate.limit</name>
+  <name>fs.azure.io.rate.limit</name>
   <value>1000</value>
 </property>
+
+<property>
+  <name>mapreduce.manifest.committer.cleanup.parallel.delete.base.first</name>
+  <value>true</value>
+</property>
+
 ```
 
 And optional settings for debugging/performance analysis
@@ -510,6 +524,7 @@ And optional settings for debugging/performance analysis
 ```
 spark.hadoop.mapreduce.outputcommitter.factory.scheme.abfs org.apache.hadoop.fs.azurebfs.commit.AzureManifestCommitterFactory
 spark.hadoop.fs.azure.io.rate.limit 1000
+spark.hadoop.mapreduce.manifest.committer.cleanup.parallel.delete.base.first true
 spark.sql.parquet.output.committer.class org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter
 spark.sql.sources.commitProtocolClass org.apache.spark.internal.io.cloud.PathOutputCommitProtocol
 

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

@@ -167,6 +167,12 @@ public abstract class AbstractManifestCommitterTest
 
   private static final int MAX_LEN = 64_000;
 
+  /**
+   * How many attempts to save manifests before giving up.
+   * Kept small to reduce sleep times and network delays.
+   */
+  public static final int SAVE_ATTEMPTS = 3;
+
   /**
    * Submitter for tasks; may be null.
    */
@@ -799,6 +805,7 @@ protected StageConfig createStageConfig(
         .withJobAttemptNumber(jobAttemptNumber)
         .withJobDirectories(attemptDirs)
         .withName(String.format(NAME_FORMAT_JOB_ATTEMPT, jobId))
+        .withManifestSaveAttempts(SAVE_ATTEMPTS)
         .withOperations(getStoreOperations())
         .withProgressable(getProgressCounter())
         .withSuccessMarkerFileLimit(100_000)