HADOOP-15469. S3A directory committer commit job fails if _temporary directory created under dest.

Contributed by Steve Loughran.

(cherry picked from commit 170f1040d46f9b1a084e6637def91e9864446acc)

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/staging/DirectoryStagingCommitter.java

@@ -93,11 +93,8 @@ public class DirectoryStagingCommitter extends StagingCommitter {
     Configuration fsConf = fs.getConf();
     switch (getConflictResolutionMode(context, fsConf)) {
     case FAIL:
-      // this was checked in setupJob, but this avoids some cases where
-      // output was created while the job was processing
-      if (fs.exists(outputPath)) {
-        throw new PathExistsException(outputPath.toString(), E_DEST_EXISTS);
-      }
+      // this was checked in setupJob; temporary files may have been
+      // created, so do not check again.
       break;
     case APPEND:
       // do nothing

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/TestStagingDirectoryOutputCommitter.java

@@ -67,15 +67,14 @@ public class TestStagingDirectoryOutputCommitter
     pathExists(mockS3, OUTPUT_PATH);
     final DirectoryStagingCommitter committer = newJobCommitter();
 
+    // this should fail
     intercept(PathExistsException.class,
         InternalCommitterConstants.E_DEST_EXISTS,
         "Should throw an exception because the path exists",
         () -> committer.setupJob(getJob()));
 
-    intercept(PathExistsException.class,
-        InternalCommitterConstants.E_DEST_EXISTS,
-        "Should throw an exception because the path exists",
-        () -> committer.commitJob(getJob()));
+    // but there are no checks in job commit (HADOOP-15469)
+    committer.commitJob(getJob());
 
     reset(mockS3);
     pathDoesNotExist(mockS3, OUTPUT_PATH);
@@ -87,7 +86,6 @@ public class TestStagingDirectoryOutputCommitter
     reset(mockS3);
     pathDoesNotExist(mockS3, OUTPUT_PATH);
     committer.commitJob(getJob());
-    verifyExistenceChecked(mockS3, OUTPUT_PATH);
     verifyCompletion(mockS3);
   }