updates getObjectMetadata, putObject & copyObject operations

Author: ahmarsuhail

hadoop-project/pom.xml

@@ -1128,6 +1128,11 @@
           </exclusion>
         </exclusions>
       </dependency>
+      <dependency>
+        <groupId>software.amazon.awssdk</groupId>
+        <artifactId>s3-transfer-manager</artifactId>
+        <version>2.17.196-PREVIEW</version>
+      </dependency>
       <dependency>
         <groupId>org.apache.mina</groupId>
         <artifactId>mina-core</artifactId>

hadoop-tools/hadoop-aws/pom.xml

@@ -497,6 +497,11 @@
       <artifactId>bundle</artifactId>
       <scope>compile</scope>
     </dependency>
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>s3-transfer-manager</artifactId>
+      <scope>compile</scope>
+    </dependency>
     <dependency>
       <groupId>org.assertj</groupId>
       <artifactId>assertj-core</artifactId>

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

@@ -31,10 +31,10 @@
 import software.amazon.awssdk.core.retry.RetryPolicy;
 import software.amazon.awssdk.http.apache.ApacheHttpClient;
 import software.amazon.awssdk.http.apache.ProxyConfiguration;
+import software.amazon.awssdk.thirdparty.org.apache.http.client.utils.URIBuilder;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.util.VersionInfo;
-import org.apache.http.client.utils.URIBuilder;
 
 import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_ESTABLISH_TIMEOUT;
 import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_MAXIMUM_CONNECTIONS;

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

@@ -28,6 +28,8 @@
 import com.amazonaws.SdkBaseException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.awscore.exception.AwsServiceException;
+import software.amazon.awssdk.core.exception.SdkException;
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -120,8 +122,13 @@ public static <T> T once(String action, String path,
       throws IOException {
     try (DurationInfo ignored = new DurationInfo(LOG, false, "%s", action)) {
       return operation.apply();
-    } catch (AmazonClientException e) {
-      throw S3AUtils.translateException(action, path, e);
+    } catch (AmazonClientException | AwsServiceException e) {
+      // TODO: This is temporary, and will be updated during error translation work.
+      if (e instanceof  AmazonClientException) {
+        throw S3AUtils.translateException(action, path, (SdkBaseException) e);
+      } else {
+        throw S3AUtils.translateExceptionV2(action, path, (SdkException) e);
+      }
     }
   }
 
@@ -466,19 +473,24 @@ public <T> T retryUntranslated(
         }
         // execute the operation, returning if successful
         return operation.apply();
-      } catch (IOException | SdkBaseException e) {
+      } catch (IOException | SdkBaseException | AwsServiceException e) {
         caught = e;
       }
       // you only get here if the operation didn't complete
       // normally, hence caught != null
 
       // translate the exception into an IOE for the retry logic
       IOException translated;
+      // TODO: Update during error translation work. This is a temporary fix to allow
+      //  getObjectMetadata to throw FNFE.
       if (caught instanceof IOException) {
         translated = (IOException) caught;
-      } else {
+      } else if (caught instanceof SdkBaseException) {
         translated = S3AUtils.translateException(text, "",
             (SdkBaseException)caught);
+      } else {
+        translated = S3AUtils.translateExceptionV2(text, "",
+            (AwsServiceException)caught);
       }
 
       try {
@@ -516,8 +528,11 @@ public <T> T retryUntranslated(
 
     if (caught instanceof IOException) {
       throw (IOException) caught;
-    } else {
+      // TODO: This is temporary, and will be updated during error translation work.
+    } else if (caught instanceof SdkBaseException) {
       throw (SdkBaseException) caught;
+    } else {
+      throw (AwsServiceException) caught;
     }
   }
 

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

@@ -18,59 +18,56 @@
 
 package org.apache.hadoop.fs.s3a;
 
-import com.amazonaws.event.ProgressEvent;
-import com.amazonaws.event.ProgressEventType;
-import com.amazonaws.event.ProgressListener;
-import com.amazonaws.services.s3.transfer.Upload;
-import org.apache.hadoop.util.Progressable;
 import org.slf4j.Logger;
 
-import static com.amazonaws.event.ProgressEventType.TRANSFER_COMPLETED_EVENT;
-import static com.amazonaws.event.ProgressEventType.TRANSFER_PART_STARTED_EVENT;
+import software.amazon.awssdk.transfer.s3.ObjectTransfer;
+import software.amazon.awssdk.transfer.s3.progress.TransferListener;
+
+import org.apache.hadoop.util.Progressable;
 
 /**
  * Listener to progress from AWS regarding transfers.
  */
-public class ProgressableProgressListener implements ProgressListener {
+public class ProgressableProgressListener implements TransferListener {
   private static final Logger LOG = S3AFileSystem.LOG;
   private final S3AFileSystem fs;
   private final String key;
   private final Progressable progress;
   private long lastBytesTransferred;
-  private final Upload upload;
 
   /**
    * Instantiate.
    * @param fs filesystem: will be invoked with statistics updates
    * @param key key for the upload
-   * @param upload source of events
    * @param progress optional callback for progress.
    */
   public ProgressableProgressListener(S3AFileSystem fs,
       String key,
-      Upload upload,
       Progressable progress) {
     this.fs = fs;
     this.key = key;
-    this.upload = upload;
     this.progress = progress;
     this.lastBytesTransferred = 0;
   }
 
   @Override
-  public void progressChanged(ProgressEvent progressEvent) {
-    if (progress != null) {
-      progress.progress();
-    }
+  public void  transferInitiated(TransferListener.Context.TransferInitiated context) {
+    fs.incrementWriteOperations();
+  }
 
-    // There are 3 http ops here, but this should be close enough for now
-    ProgressEventType pet = progressEvent.getEventType();
-    if (pet == TRANSFER_PART_STARTED_EVENT ||
-        pet == TRANSFER_COMPLETED_EVENT) {
-      fs.incrementWriteOperations();
+  @Override
+  public void transferComplete(TransferListener.Context.TransferComplete context) {
+    fs.incrementWriteOperations();
+  }
+
+  @Override
+  public void bytesTransferred(TransferListener.Context.BytesTransferred context) {
+
+    if(progress != null) {
+      progress.progress();
     }
 
-    long transferred = upload.getProgress().getBytesTransferred();
+    long transferred = context.progressSnapshot().bytesTransferred();
     long delta = transferred - lastBytesTransferred;
     fs.incrementPutProgressStatistics(key, delta);
     lastBytesTransferred = transferred;
@@ -81,9 +78,10 @@ public void progressChanged(ProgressEvent progressEvent) {
    * This can handle race conditions in setup/teardown.
    * @return the number of bytes which were transferred after the notification
    */
-  public long uploadCompleted() {
-    long delta = upload.getProgress().getBytesTransferred() -
-        lastBytesTransferred;
+  public long uploadCompleted(ObjectTransfer upload) {
+
+    long delta =
+        upload.progress().snapshot().bytesTransferred() - lastBytesTransferred;
     if (delta > 0) {
       LOG.debug("S3A write delta changed after finished: {} bytes", delta);
       fs.incrementPutProgressStatistics(key, delta);

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

@@ -36,8 +36,6 @@
 import com.amazonaws.event.ProgressEventType;
 import com.amazonaws.event.ProgressListener;
 import com.amazonaws.services.s3.model.PartETag;
-import com.amazonaws.services.s3.model.PutObjectRequest;
-import com.amazonaws.services.s3.model.PutObjectResult;
 import com.amazonaws.services.s3.model.UploadPartRequest;
 
 import org.apache.hadoop.fs.s3a.impl.PutObjectOptions;
@@ -50,6 +48,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.PutObjectResponse;
+
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.fs.Abortable;
@@ -575,24 +576,30 @@ private int putObject() throws IOException {
     final PutObjectRequest putObjectRequest = uploadData.hasFile() ?
         writeOperationHelper.createPutObjectRequest(
             key,
-            uploadData.getFile(),
-            builder.putOptions)
+            uploadData.getFile().length(),
+            builder.putOptions,
+            true)
         : writeOperationHelper.createPutObjectRequest(
             key,
-            uploadData.getUploadStream(),
             size,
-            builder.putOptions);
-    BlockUploadProgress callback =
-        new BlockUploadProgress(
-            block, progressListener,  now());
-    putObjectRequest.setGeneralProgressListener(callback);
+            builder.putOptions,
+        false);
+
+          // TODO: You cannot currently add progress listeners to requests not via the TM.
+          // There is an open ticket for this with the SDK team. But need to check how important
+          // this is for us?
+    //    BlockUploadProgress callback =
+    //        new BlockUploadProgress(
+    //            block, progressListener,  now());
+    // putObjectRequest.setGeneralProgressListener(callback);
     statistics.blockUploadQueued(size);
-    ListenableFuture<PutObjectResult> putObjectResult =
+    ListenableFuture<PutObjectResponse> putObjectResult =
         executorService.submit(() -> {
           try {
             // the putObject call automatically closes the input
             // stream afterwards.
-            return writeOperationHelper.putObject(putObjectRequest, builder.putOptions);
+            return writeOperationHelper.putObject(putObjectRequest, builder.putOptions, uploadData,
+                uploadData.hasFile());
           } finally {
             cleanupWithLogger(LOG, uploadData, block);
           }

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

@@ -47,7 +47,7 @@
  * Set of classes to support output streaming into blocks which are then
  * uploaded as to S3 as a single PUT, or as part of a multipart request.
  */
-final class S3ADataBlocks {
+public final class S3ADataBlocks {
 
   private static final Logger LOG =
       LoggerFactory.getLogger(S3ADataBlocks.class);
@@ -101,15 +101,15 @@ static BlockFactory createFactory(S3AFileSystem owner,
    * It can be one of a file or an input stream.
    * When closed, any stream is closed. Any source file is untouched.
    */
-  static final class BlockUploadData implements Closeable {
+public static final class BlockUploadData implements Closeable {
     private final File file;
     private final InputStream uploadStream;
 
     /**
      * File constructor; input stream will be null.
      * @param file file to upload
      */
-    BlockUploadData(File file) {
+   public BlockUploadData(File file) {
       Preconditions.checkArgument(file.exists(), "No file: " + file);
       this.file = file;
       this.uploadStream = null;
@@ -119,7 +119,7 @@ static final class BlockUploadData implements Closeable {
      * Stream constructor, file field will be null.
      * @param uploadStream stream to upload
      */
-    BlockUploadData(InputStream uploadStream) {
+   public BlockUploadData(InputStream uploadStream) {
       Preconditions.checkNotNull(uploadStream, "rawUploadStream");
       this.uploadStream = uploadStream;
       this.file = null;