HADOOP-17318. Support concurrent S3A jobs with conflicting app attempt IDs

    * ITests for the uuid generate/require logic
    * Pendingset and _SUCCESS files add jobId field
    * And on job commit, the jobID of pendingset files are validated

    The validation will detect and fail if a job with a different ID
    has got a .pendingset file into the directory used by the current
    job.

    The _SUCCESS file is to aid auditing/testing

    Change-Id: I07a6a2d00ac5598c8f961aebbc9c9fdbb70ab51a

commit f2128bd1bf94de36ee1e3ed542de0a0e839307e3
Author: Steve Loughran <[email protected]>
Date:   Tue Oct 27 17:05:12 2020 +0000

    HADOOP-17318. Support concurrent S3A jobs with conflicting app attempt IDs.

    Have a job UUID everywhere, which for spark must be passed in or
    self-generated; for MR it will be the yarn app attempt.

    The task attempts are still used under this where task work needs
    to be differentiated.

    Examples
    * temp dir for staging
    * magic path for magic committers
    * HDFS dir for staging summary info

    Change-Id: I17c641280d916ea1ad4ce4407215d07e488954af

commit 0ae79187d6c821eef13872e4c384948729b9c72d
Author: Steve Loughran <[email protected]>
Date:   Tue Oct 20 21:36:36 2020 +0100

    HADOOP-17318. Support concurrent S3A commit jobs slightly better.

    All S3A committers can have purging pending deletes on job commit
    disabled. (new option; old one deprecated).

    More logging of what is going on with individual file load/upload/commit
    (including with duration)

    Test of concurrent jobs designed to trigger the specific failure conditions
    of Staging (job2 commit after task1 commit) and Magic (job2 commit before
    task2 commit)

    Change-Id: If560c7541c287dc6d4c2f1af395c93b838495139

Change-Id: I2374b904bfb65399e08084e6c2b78237ec1603cd

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/resources/core-default.xml

@@ -1925,20 +1925,13 @@
 </property>
 
 <property>
-  <name>fs.s3a.committer.staging.abort.pending.uploads</name>
+  <name>fs.s3a.committer.abort.pending.uploads</name>
   <value>true</value>
   <description>
-    Should the staging committers abort all pending uploads to the destination
+    Should the committers abort all pending uploads to the destination
     directory?
 
-    Changing this if more than one partitioned committer is
-    writing to the same destination tree simultaneously; otherwise
-    the first job to complete will cancel all outstanding uploads from the
-    others. However, it may lead to leaked outstanding uploads from failed
-    tasks. If disabled, configure the bucket lifecycle to remove uploads
-    after a time period, and/or set up a workflow to explicitly delete
-    entries. Otherwise there is a risk that uncommitted uploads may run up
-    bills.
+    Set to false if more than one job is writing to the same directory tree.
   </description>
 </property>
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -180,6 +180,8 @@
 import static org.apache.hadoop.fs.s3a.auth.RolePolicies.allowS3Operations;
 import static org.apache.hadoop.fs.s3a.auth.delegation.S3ADelegationTokens.TokenIssuingPolicy.NoTokensAvailable;
 import static org.apache.hadoop.fs.s3a.auth.delegation.S3ADelegationTokens.hasDelegationTokenBinding;
+import static org.apache.hadoop.fs.s3a.commit.CommitConstants.FS_S3A_COMMITTER_ABORT_PENDING_UPLOADS;
+import static org.apache.hadoop.fs.s3a.commit.CommitConstants.FS_S3A_COMMITTER_STAGING_ABORT_PENDING_UPLOADS;
 import static org.apache.hadoop.fs.s3a.impl.CallableSupplier.submit;
 import static org.apache.hadoop.fs.s3a.impl.CallableSupplier.waitForCompletionIgnoringExceptions;
 import static org.apache.hadoop.fs.s3a.impl.ErrorTranslation.isUnknownBucket;
@@ -314,9 +316,10 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
   /** Add any deprecated keys. */
   @SuppressWarnings("deprecation")
   private static void addDeprecatedKeys() {
-    // this is retained as a placeholder for when new deprecated keys
-    // need to be added.
     Configuration.DeprecationDelta[] deltas = {
+        new Configuration.DeprecationDelta(
+            FS_S3A_COMMITTER_STAGING_ABORT_PENDING_UPLOADS,
+            FS_S3A_COMMITTER_ABORT_PENDING_UPLOADS)
     };
 
     if (deltas.length > 0) {
@@ -4581,7 +4584,7 @@ public List<MultipartUpload> listMultipartUploads(String prefix)
    */
   @Retries.OnceRaw
   void abortMultipartUpload(String destKey, String uploadId) {
-    LOG.debug("Aborting multipart upload {} to {}", uploadId, destKey);
+    LOG.info("Aborting multipart upload {} to {}", uploadId, destKey);
     getAmazonS3Client().abortMultipartUpload(
         new AbortMultipartUploadRequest(getBucket(),
             destKey,

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/WriteOperationHelper.java

@@ -131,6 +131,8 @@ protected WriteOperationHelper(S3AFileSystem owner, Configuration conf) {
    */
   void operationRetried(String text, Exception ex, int retries,
       boolean idempotent) {
+    LOG.info("{}: Retried {}: {}", retries, text, ex.toString());
+    LOG.debug("Stack", ex);
     owner.operationRetried(text, ex, retries, idempotent);
   }
 
@@ -323,7 +325,9 @@ public CompleteMultipartUploadResult completeMPUwithRetries(
   public void abortMultipartUpload(String destKey, String uploadId,
       Retried retrying)
       throws IOException {
-    invoker.retry("Aborting multipart upload", destKey, true,
+    invoker.retry("Aborting multipart upload ID " + uploadId,
+        destKey,
+        true,
         retrying,
         () -> owner.abortMultipartUpload(
             destKey,
@@ -585,7 +589,8 @@ public BulkOperationState initiateOperation(final Path path,
   @Retries.RetryTranslated
   public UploadPartResult uploadPart(UploadPartRequest request)
       throws IOException {
-    return retry("upload part",
+    return retry("upload part #" + request.getPartNumber()
+        + " upload "+ request.getUploadId(),
         request.getKey(),
         true,
         () -> owner.uploadPart(request));