MAPREDUCE-7474. use commitFile() to rename the temporary task manifest

Move from classic rename() to commitFile() to rename the file,
after calling getFileStatus() to get its length and possibly etag.

This becomes a rename() on gcs/hdfs anyway, but on abfs it does reach
the ResilientCommitByRename callbacks in abfs, which report on
the outcome to the caller...which is then logged at WARN.

test changes to match the codepath changes, including improvements
in fault injection.

Change-Id: I757a77c8d2b7a7f1cf2ce32d109ce1baa6a90ec2

Author: steveloughran

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

@@ -676,15 +676,20 @@ protected final <T extends AbstractManifestData> T saveManifest(
           // -this is just a little bit of due diligence.
           deleteRecursive(tempPath, OP_DELETE);
 
-          // save the temp file, overwriting any which remains from an earlier attempt
+          // save the temp file.
           operations.save(manifestData, tempPath, true);
+          // get the length and etag.
+          final FileStatus st = getFileStatus(tempPath);
+
+          // commit rename of temporary file to the final path; deleting the destination first.
+          final CommitOutcome outcome = commitFile(
+              new FileEntry(tempPath, finalPath, st.getLen(), getEtag(st)),
+              true);
+          if (outcome.recovered) {
+            LOG.warn("Task manifest file {} committed using rename recovery",
+                manifestData);
+          }
 
-          // delete the destination in case it exists either from a failed previous
-          // attempt or from a concurrent task commit.
-          delete(finalPath, true, OP_DELETE);
-
-          // rename temporary file to the final path.
-          renameFile(tempPath, finalPath);
         });
         // success: save the manifest and declare success
         savedManifest = manifestData;
@@ -797,13 +802,14 @@ protected final CommitOutcome commitFile(FileEntry entry,
         // note any delay which took place
         noteAnyRateLimiting(STORE_IO_RATE_LIMITED, result.getWaitTime());
       }
+      return new CommitOutcome(result.recovered());
     } else {
       // commit with a simple rename; failures will be escalated.
       executeRenamingOperation("renameFile", source, dest,
           OP_COMMIT_FILE_RENAME, () ->
               operations.renameFile(source, dest));
+      return new CommitOutcome(false);
     }
-    return new CommitOutcome();
   }
 
   /**
@@ -923,6 +929,14 @@ private PathIOException escalateRenameFailure(String operation,
    */
   public static final class CommitOutcome {
 
+    /**
+     * Dit the commit recover from a failure?
+     */
+    public final boolean recovered;
+
+    public CommitOutcome(final boolean recovered) {
+      this.recovered = recovered;
+    }
   }
 
   /**

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

@@ -184,7 +184,8 @@ protected Result executeStage(
           if (exception == null) {
             // success: record this as the outcome,
             outcome = Outcome.DELETED;
-            // and will skip the parallel delete
+            // and flag that the the parallel delete should be skipped because the
+            // base directory is alredy deleted.
             baseDirDeleted = true;
           } else {
             // failure: log and continue

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

@@ -34,10 +34,15 @@ The protocol and its correctness are covered in
 [Manifest Committer Protocol](manifest_committer_protocol.html).
 
 It was added in March 2022.
-As of April 2024, the only problems have been scale related, rather than
-algorithm correctness.
+As of April 2024, the problems which surfaced have been
+* Memory use at scale.
+* Directory deletion scalability.
+* Resilience to task commit to rename failures.
 
-<!-- MACRO{toc|fromDepth=0|toDepth=2} -->
+That is: the core algorithms is correct, but task commit
+robustness was insufficient to some failure conditions.
+And scale is always a challenge, even with components tested through
+large TPC-DS test runs.
 
 ## Problem:
 
@@ -566,12 +571,12 @@ may surface in cloud storage.
 * General resilience to cleanup issues escalating to job failures.
 
 
-| Option | Meaning | Default Value |
-|--------|---------|---------------|
-| `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 | `false` |
+| Option                                                            | Meaning                                                            | Default Value |
+|-------------------------------------------------------------------|--------------------------------------------------------------------|---------------|
+| `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 | `false`       |
 
 The algorithm is:
 
@@ -790,12 +795,12 @@ spark.hadoop.mapreduce.manifest.committer.summary.report.directory  (optional: U
 There are some advanced options which are intended for development and testing,
 rather than production use.
 
-| Option | Meaning                                      | Default Value |
-|--------|----------------------------------------------|---------------|
-| `mapreduce.manifest.committer.manifest.save.attempts` | How many attempts should be made to commit a task manifest? | `5` |
-| `mapreduce.manifest.committer.store.operations.classname` | Classname for Manifest Store Operations      | `""`          |
-| `mapreduce.manifest.committer.validate.output` | Perform output validation?                   | `false`       |
-| `mapreduce.manifest.committer.writer.queue.capacity` | Queue capacity for writing intermediate file | `32`          |
+| Option                                                    | Meaning                                                     | Default Value |
+|-----------------------------------------------------------|-------------------------------------------------------------|---------------|
+| `mapreduce.manifest.committer.manifest.save.attempts`     | How many attempts should be made to commit a task manifest? | `5`           |
+| `mapreduce.manifest.committer.store.operations.classname` | Classname for Manifest Store Operations                     | `""`          |
+| `mapreduce.manifest.committer.validate.output`            | Perform output validation?                                  | `false`       |
+| `mapreduce.manifest.committer.writer.queue.capacity`      | Queue capacity for writing intermediate file                | `32`          |
 
 ### `mapreduce.manifest.committer.manifest.save.attempts`
 

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

@@ -52,7 +52,8 @@
 import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 
 /**
- * Test committing a task.
+ * Test committing a task, with lots of fault injection to validate
+ * resilience to transient failures.
  */
 public class TestCommitTaskStage extends AbstractManifestCommitterTest {
 
@@ -303,8 +304,8 @@ public void testManifestRenameLateTimeoutsRecovery() throws Throwable {
   }
 
   @Test
-  public void testManifestDeleteInRenameErrorHandlerFailure() throws Throwable {
-    describe("Testing failure in the delete call made during cleanup");
+  public void testFailureToDeleteManifestPath() throws Throwable {
+    describe("Testing failure in the delete call made before renaming the manifest");
 
     UnreliableManifestStoreOperations failures = makeStoreOperationsUnreliable();
     StageConfig stageConfig = createStageConfig();
@@ -315,6 +316,9 @@ public void testManifestDeleteInRenameErrorHandlerFailure() throws Throwable {
     // final manifest file is by task ID
     Path manifestFile = manifestPathForTask(manifestDir,
         stageConfig.getTaskId());
+    // put a file in as there is a check for it before the delete
+    ContractTestUtils.touch(getFileSystem(), manifestFile);
+    /* and the delete shall fail */
     failures.addDeletePathToFail(manifestFile);
     Path manifestTempFile = manifestTempPathForTaskAttempt(manifestDir,
         stageConfig.getTaskAttemptId());
@@ -332,10 +336,10 @@ public void testManifestDeleteInRenameErrorHandlerFailure() throws Throwable {
 
 
   /**
-   * Failure of delete before rename.
+   * Failure of delete before saving the manifest to a temporary path.
    */
   @Test
-  public void testFailureOfDeleteBeforeRename() throws Throwable {
+  public void testFailureOfDeleteBeforeSavingTemporaryFile() throws Throwable {
     describe("Testing failure in the delete call made before rename");
 
     UnreliableManifestStoreOperations failures = makeStoreOperationsUnreliable();
@@ -344,14 +348,11 @@ public void testFailureOfDeleteBeforeRename() throws Throwable {
     new SetupTaskStage(stageConfig).apply("setup");
 
     final Path manifestDir = stageConfig.getTaskManifestDir();
-    // final manifest file is by task ID
-    Path manifestFile = manifestPathForTask(manifestDir,
-        stageConfig.getTaskId());
     Path manifestTempFile = manifestTempPathForTaskAttempt(manifestDir,
         stageConfig.getTaskAttemptId());
 
     // delete will fail
-    failures.addDeletePathToFail(manifestFile);
+    failures.addDeletePathToFail(manifestTempFile);
 
     // first verify that if too many attempts fail, the task will fail
     failures.setFailureLimit(SAVE_ATTEMPTS + 1);

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

@@ -452,13 +452,19 @@ public boolean storeSupportsResilientCommit() {
   @Override
   public CommitFileResult commitFile(final FileEntry entry)
       throws IOException {
+    final String op = "commitFile";
+    final Path source = entry.getSourcePath();
+    maybeTimeout(op, source, renamePathsToTimeoutBeforeRename);
     if (renameToFailWithException) {
-      maybeRaiseIOE("commitFile",
-          entry.getSourcePath(), renameSourceFilesToFail);
-      maybeRaiseIOE("commitFile",
+      maybeRaiseIOE(op,
+          source, renameSourceFilesToFail);
+      maybeRaiseIOE(op,
           entry.getDestPath().getParent(), renameDestDirsToFail);
     }
-    return wrappedOperations.commitFile(entry);
+    final CommitFileResult result = wrappedOperations.commitFile(entry);
+    // post rename timeout.
+    maybeTimeout(op, source, renamePathsToTimeoutAfterRename);
+    return result;
   }
 
   @Override

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/commit/AbfsCommitTestHelper.java

@@ -23,6 +23,7 @@
 import org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterConstants;
 
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_READ_SMALL_FILES_COMPLETELY;
+import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterConstants.OPT_CLEANUP_PARALLEL_DELETE_BASE_FIRST;
 import static org.apache.hadoop.mapreduce.lib.output.committer.manifest.ManifestCommitterConstants.OPT_STORE_OPERATIONS_CLASS;
 
 /**
@@ -51,9 +52,10 @@ static Configuration prepareTestConfiguration(
     final String size = Integer.toString(192);
     conf.setIfUnset(ManifestCommitterConstants.OPT_IO_PROCESSORS, size);
     conf.setIfUnset(ManifestCommitterConstants.OPT_WRITER_QUEUE_CAPACITY, size);
-    // no need for parallel delete here as we aren't at the scale where unified delete
-    // is going to time out
-    conf.setBooleanIfUnset(ManifestCommitterConstants.OPT_CLEANUP_PARALLEL_DELETE, false);
+    // enable parallel delete but ask for base deletion first,
+    // which is now our recommended azure option
+    conf.setBoolean(ManifestCommitterConstants.OPT_CLEANUP_PARALLEL_DELETE, true);
+    conf.setBoolean(OPT_CLEANUP_PARALLEL_DELETE_BASE_FIRST, true);
 
     return conf;
   }