Merge pull request apache#8 from ahmarsuhail/HADOOP-18073-sdk-upgrade-mpu-operations

Hadoop 18073 sdk upgrade mpu operations

Author: ahmarsuhail

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

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a;
+
+/**
+ * Enum to map AWS SDK V1 Acl values to SDK V2.
+ */
+public enum AWSCannedACL {
+  Private("PRIVATE"),
+  PublicRead("PUBLIC_READ"),
+  PublicReadWrite("PUBLIC_READ_WRITE"),
+  AuthenticatedRead("AUTHENTICATED_READ"),
+  AwsExecRead("AWS_EXEC_READ"),
+  BucketOwnerRead("BUCKET_OWNER_READ"),
+  BucketOwnerFullControl("BUCKET_OWNER_FULL_CONTROL");
+
+  private final String value;
+
+  AWSCannedACL(String value){
+    this.value = value;
+  }
+
+  public String toString() { return this.value; }
+}

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

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a;
+
+import java.util.List;
+
+import software.amazon.awssdk.services.s3.model.S3Error;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Exception raised in {@link S3AFileSystem#deleteObjects} when
+ * one or more of the keys could not be deleted.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class MultiObjectDeleteException extends RuntimeException {
+
+  private final List<S3Error> errors;
+
+  public MultiObjectDeleteException(List<S3Error> errors) {
+    super(errors.toString());
+    this.errors = errors;
+  }
+
+  public List<S3Error> errors() { return errors; }
+}

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

@@ -23,13 +23,14 @@
 import java.util.NoSuchElementException;
 import javax.annotation.Nullable;
 
-import com.amazonaws.services.s3.AmazonS3;
-import com.amazonaws.services.s3.model.ListMultipartUploadsRequest;
-import com.amazonaws.services.s3.model.MultipartUpload;
-import com.amazonaws.services.s3.model.MultipartUploadListing;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.ListMultipartUploadsRequest;
+import software.amazon.awssdk.services.s3.model.ListMultipartUploadsResponse;
+import software.amazon.awssdk.services.s3.model.MultipartUpload;
+
 import org.apache.hadoop.fs.RemoteIterator;
 import org.apache.hadoop.fs.s3a.api.RequestFactory;
 import org.apache.hadoop.fs.s3a.impl.StoreContext;
@@ -43,7 +44,7 @@
  * MultipartUtils upload-specific functions for use by S3AFileSystem and Hadoop
  * CLI.
  * The Audit span active when
- * {@link #listMultipartUploads(StoreContext, AmazonS3, String, int)}
+ * {@link #listMultipartUploads(StoreContext, S3Client, String, int)}
  * was invoked is retained for all subsequent operations.
  */
 public final class MultipartUtils {
@@ -67,7 +68,7 @@ private MultipartUtils() { }
    */
   static MultipartUtils.UploadIterator listMultipartUploads(
       final StoreContext storeContext,
-      AmazonS3 s3,
+      S3Client s3,
       @Nullable String prefix,
       int maxKeys)
       throws IOException {
@@ -84,14 +85,14 @@ static MultipartUtils.UploadIterator listMultipartUploads(
    * at the time the iterator was constructed.
    */
   static class ListingIterator implements
-      RemoteIterator<MultipartUploadListing> {
+      RemoteIterator<ListMultipartUploadsResponse> {
 
     private final String prefix;
 
     private final RequestFactory requestFactory;
 
     private final int maxKeys;
-    private final AmazonS3 s3;
+    private final S3Client s3;
     private final Invoker invoker;
 
     private final AuditSpan auditSpan;
@@ -101,7 +102,7 @@ static class ListingIterator implements
     /**
      * Most recent listing results.
      */
-    private MultipartUploadListing listing;
+    private ListMultipartUploadsResponse listing;
 
     /**
      * Indicator that this is the first listing.
@@ -114,7 +115,7 @@ static class ListingIterator implements
     private int listCount = 0;
 
     ListingIterator(final StoreContext storeContext,
-        AmazonS3 s3,
+        S3Client s3,
         @Nullable String prefix,
         int maxKeys) throws IOException {
       this.storeContext = storeContext;
@@ -153,7 +154,7 @@ public boolean hasNext() throws IOException {
      */
     @Override
     @Retries.RetryTranslated
-    public MultipartUploadListing next() throws IOException {
+    public ListMultipartUploadsResponse next() throws IOException {
       if (firstListing) {
         firstListing = false;
       } else {
@@ -171,32 +172,34 @@ public MultipartUploadListing next() throws IOException {
     public String toString() {
       return "Upload iterator: prefix " + prefix
           + "; list count " + listCount
-          + "; upload count " + listing.getMultipartUploads().size()
+          + "; upload count " + listing.uploads().size()
           + "; isTruncated=" + listing.isTruncated();
     }
 
     @Retries.RetryTranslated
     private void requestNextBatch() throws IOException {
       try (AuditSpan span = auditSpan.activate()) {
-        ListMultipartUploadsRequest req = requestFactory
-            .newListMultipartUploadsRequest(prefix);
+        ListMultipartUploadsRequest.Builder requestBuilder = requestFactory
+            .newListMultipartUploadsRequestBuilder(prefix);
         if (!firstListing) {
-          req.setKeyMarker(listing.getNextKeyMarker());
-          req.setUploadIdMarker(listing.getNextUploadIdMarker());
+          requestBuilder.keyMarker(listing.nextKeyMarker());
+          requestBuilder.uploadIdMarker(listing.nextUploadIdMarker());
         }
-        req.setMaxUploads(maxKeys);
+        requestBuilder.maxUploads(maxKeys);
+
+        ListMultipartUploadsRequest request = requestBuilder.build();
 
         LOG.debug("[{}], Requesting next {} uploads prefix {}, " +
             "next key {}, next upload id {}", listCount, maxKeys, prefix,
-            req.getKeyMarker(), req.getUploadIdMarker());
+            request.keyMarker(), request.uploadIdMarker());
         listCount++;
 
         listing = invoker.retry("listMultipartUploads", prefix, true,
             trackDurationOfOperation(storeContext.getInstrumentation(),
                 MULTIPART_UPLOAD_LIST.getSymbol(),
-                () -> s3.listMultipartUploads(req)));
+                () -> s3.listMultipartUploads(requestBuilder.build())));
         LOG.debug("Listing found {} upload(s)",
-            listing.getMultipartUploads().size());
+            listing.uploads().size());
         LOG.debug("New listing state: {}", this);
       }
     }
@@ -216,14 +219,14 @@ public static class UploadIterator
      */
     private ListingIterator lister;
     /** Current listing: the last upload listing we fetched. */
-    private MultipartUploadListing listing;
+    private ListMultipartUploadsResponse listing;
     /** Iterator over the current listing. */
     private ListIterator<MultipartUpload> batchIterator;
 
     @Retries.RetryTranslated
     public UploadIterator(
         final StoreContext storeContext,
-        AmazonS3 s3,
+        S3Client s3,
         int maxKeys,
         @Nullable String prefix)
         throws IOException {
@@ -249,7 +252,7 @@ public MultipartUpload next() throws IOException {
     private boolean requestNextBatch() throws IOException {
       if (lister.hasNext()) {
         listing = lister.next();
-        batchIterator = listing.getMultipartUploads().listIterator();
+        batchIterator = listing.uploads().listIterator();
         return batchIterator.hasNext();
       }
       return false;

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

@@ -48,6 +48,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import software.amazon.awssdk.services.s3.model.CompletedPart;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.services.s3.model.PutObjectResponse;
 
@@ -419,7 +420,7 @@ public void close() throws IOException {
           uploadCurrentBlock(true);
         }
         // wait for the partial uploads to finish
-        final List<PartETag> partETags =
+        final List<CompletedPart> partETags =
             multiPartUpload.waitForAllPartUploads();
         bytes = bytesSubmitted;
 
@@ -742,7 +743,7 @@ protected IOStatisticsAggregator getThreadIOStatistics() {
    */
   private class MultiPartUpload {
     private final String uploadId;
-    private final List<ListenableFuture<PartETag>> partETagsFutures;
+    private final List<ListenableFuture<CompletedPart>> partETagsFutures;
     private int partsSubmitted;
     private int partsUploaded;
     private long bytesSubmitted;
@@ -874,20 +875,22 @@ private void uploadBlockAsync(final S3ADataBlocks.DataBlock block,
               block, progressListener, now());
       request.setGeneralProgressListener(callback);
       statistics.blockUploadQueued(block.dataSize());
-      ListenableFuture<PartETag> partETagFuture =
+      ListenableFuture<CompletedPart> partETagFuture =
           executorService.submit(() -> {
             // this is the queued upload operation
             // do the upload
             try {
               LOG.debug("Uploading part {} for id '{}'",
                   currentPartNumber, uploadId);
+              // TODO: This needs to updated during uploadPart work. Remove PartEtag import.
               PartETag partETag = writeOperationHelper.uploadPart(request)
                   .getPartETag();
               LOG.debug("Completed upload of {} to part {}",
                   block, partETag.getETag());
               LOG.debug("Stream statistics of {}", statistics);
               partsUploaded++;
-              return partETag;
+              return CompletedPart.builder().eTag(partETag.getETag()).partNumber(currentPartNumber)
+                  .build();
             } catch (IOException e) {
               // save immediately.
               noteUploadFailure(e);
@@ -905,7 +908,7 @@ private void uploadBlockAsync(final S3ADataBlocks.DataBlock block,
      * @return list of results
      * @throws IOException IO Problems
      */
-    private List<PartETag> waitForAllPartUploads() throws IOException {
+    private List<CompletedPart> waitForAllPartUploads() throws IOException {
       LOG.debug("Waiting for {} uploads to complete", partETagsFutures.size());
       try {
         return Futures.allAsList(partETagsFutures).get();
@@ -929,7 +932,7 @@ private List<PartETag> waitForAllPartUploads() throws IOException {
      */
     private void cancelAllActiveFutures() {
       LOG.debug("Cancelling futures");
-      for (ListenableFuture<PartETag> future : partETagsFutures) {
+      for (ListenableFuture<CompletedPart> future : partETagsFutures) {
         future.cancel(true);
       }
     }
@@ -941,7 +944,7 @@ private void cancelAllActiveFutures() {
      * @param partETags list of partial uploads
      * @throws IOException on any problem
      */
-    private void complete(List<PartETag> partETags)
+    private void complete(List<CompletedPart> partETags)
         throws IOException {
       maybeRethrowUploadFailure();
       AtomicInteger errorCount = new AtomicInteger(0);