HDFS-13934. S3A MPUploader uses bulk operation state to avoid

excessive DDB IO.

Change-Id: Icbab3ef1e93ad3da87a3b29d11698213d1f0e26c

Author: steveloughran

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

@@ -254,7 +254,7 @@ private CompleteMultipartUploadResult finalizeMultipartUpload(
           "No upload parts in multipart upload to " + destKey);
     }
     CompleteMultipartUploadResult uploadResult =
-        invoker.retry("Completing multipart commit", destKey,
+        invoker.retry("Completing multipart upload", destKey,
             true,
             retrying,
             () -> {
@@ -539,8 +539,20 @@ public CompleteMultipartUploadResult commitUpload(
    */
   public BulkOperationState initiateCommitOperation(
       Path path) throws IOException {
+    return initiateOperation(path, BulkOperationState.OperationType.Commit);
+  }
+
+  /**
+   * Initiate a commit operation through any metastore.
+   * @param path path under which the writes will all take place.
+   * @param operationType operation to initiate
+   * @return an possibly null operation state from the metastore.
+   * @throws IOException failure to instantiate.
+   */
+  public BulkOperationState initiateOperation(final Path path,
+      final BulkOperationState.OperationType operationType) throws IOException {
     return S3Guard.initiateBulkWrite(owner.getMetadataStore(),
-        BulkOperationState.OperationType.Commit, path);
+        operationType, path);
   }
 
   /**

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

@@ -30,7 +30,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import com.amazonaws.services.s3.model.CompleteMultipartUploadResult;
 import com.amazonaws.services.s3.model.PartETag;
@@ -52,6 +51,7 @@
 import org.apache.hadoop.fs.UploadHandle;
 import org.apache.hadoop.fs.impl.AbstractMultipartUploader;
 import org.apache.hadoop.fs.s3a.WriteOperationHelper;
+import org.apache.hadoop.fs.s3a.s3guard.BulkOperationState;
 
 /**
  * MultipartUploader for S3AFileSystem. This uses the S3 multipart
@@ -71,15 +71,40 @@ public class S3AMultipartUploader extends AbstractMultipartUploader {
 
   private final StoreContext context;
 
+  private final Path path;
+
+  /**
+   * Bulk state -demand created during completion operations.
+   */
+  private BulkOperationState operationState;
+
   public S3AMultipartUploader(final S3AMultipartUploaderBuilder builder,
       final WriteOperationHelper writeHelper,
       final StoreContext context) {
     this.builder = builder;
     this.writeHelper = writeHelper;
+    path = builder.getPath();
 
     this.context = context;
   }
 
+  @Override
+  public void close() throws IOException {
+    if (operationState != null) {
+      operationState.close();
+    }
+    super.close();
+  }
+
+  private synchronized BulkOperationState retrieveOperationState()
+      throws IOException {
+    if (operationState == null) {
+      operationState = writeHelper.initiateOperation(path,
+          BulkOperationState.OperationType.Upload);
+    }
+    return operationState;
+  }
+
   @Override
   public CompletableFuture<UploadHandle> initialize(Path filePath)
       throws IOException {
@@ -142,13 +167,14 @@ public CompletableFuture<PathHandle> complete(Path filePath,
       totalLength += result.getLeft();
       eTags.add(new PartETag(handle.getKey(), result.getRight()));
     }
-    AtomicInteger errorCount = new AtomicInteger(0);
+    // retrieve/create operation state for scalability of completion.
+    final BulkOperationState state = retrieveOperationState();
     long finalLen = totalLength;
     return context.submit(new CompletableFuture<>(),
         () -> {
           CompleteMultipartUploadResult result
-              = writeHelper.completeMPUwithRetries(
-              key, uploadIdStr, eTags, finalLen, errorCount);
+              = writeHelper.commitUpload(
+              key, uploadIdStr, eTags, finalLen, state);
 
           byte[] eTag = result.getETag().getBytes(Charsets.UTF_8);
           return (PathHandle) () -> ByteBuffer.wrap(eTag);
@@ -170,7 +196,6 @@ public CompletableFuture<Void> abort(Path filePath, UploadHandle uploadId)
         });
   }
 
-
   /**
    * Build the payload for marshalling.
    * @param eTag upload etag

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

@@ -102,5 +102,9 @@ public enum OperationType {
      * Mkdir operation.
      */
     Mkdir,
+    /**
+     * Multipart upload operation.
+     */
+    Upload
   }
 }