HADOOP-17833. Rebase and add provisional awareness of storage class

rebase onto trunk with  HADOOP-12020/storage class; use PutObjectOptions
as the place to pass the option from createFile() to the requests.

The feature itself is not implemented, just prepared for.

Change-Id: I5d2e57e672f49021c3cb8bfa3b1a3daf09c61a38

Author: steveloughran

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

@@ -559,12 +559,12 @@ private int putObject() throws IOException {
         writeOperationHelper.createPutObjectRequest(
             key,
             uploadData.getFile(),
-            builder.putOptions.getHeaders())
+            builder.putOptions)
         : writeOperationHelper.createPutObjectRequest(
             key,
             uploadData.getUploadStream(),
             size,
-            builder.putOptions.getHeaders());
+            builder.putOptions);
     BlockUploadProgress callback =
         new BlockUploadProgress(
             block, progressListener,  now());

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

@@ -1691,11 +1691,11 @@ private FSDataOutputStream innerCreateFile(
         committerIntegration.createTracker(path, key, outputStreamStatistics);
     String destKey = putTracker.getDestKey();
 
-    // put options are from the path, unless performance is requested,
-    // where it is always keeping dirs.
+    // put options are derived from the path and the
+    // option builder.
     boolean keep = performance || keepDirectoryMarkers(path);
     final PutObjectOptions putOptions =
-        new PutObjectOptions(keep, options.getHeaders());
+        new PutObjectOptions(keep, null, options.getHeaders());
 
     final S3ABlockOutputStream.BlockOutputStreamBuilder builder =
         S3ABlockOutputStream.builder()
@@ -2739,7 +2739,7 @@ private DeleteObjectsResult deleteObjects(DeleteObjectsRequest deleteRequest)
    */
   public PutObjectRequest newPutObjectRequest(String key,
       ObjectMetadata metadata, File srcfile) {
-    return requestFactory.newPutObjectRequest(key, metadata, srcfile);
+    return requestFactory.newPutObjectRequest(key, metadata, null, srcfile);
   }
 
   /**

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

@@ -24,7 +24,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import com.amazonaws.services.s3.model.AmazonS3Exception;
@@ -234,49 +233,47 @@ private void deactivateAuditSpan() {
    * @param destKey destination key
    * @param inputStream source data.
    * @param length size, if known. Use -1 for not known
-   * @param headers optional map of custom headers.
+   * @param options options for the request
    * @return the request
    */
   @Retries.OnceRaw
   public PutObjectRequest createPutObjectRequest(String destKey,
       InputStream inputStream,
       long length,
-      @Nullable final Map<String, String> headers) {
+      final PutObjectOptions options) {
     activateAuditSpan();
     ObjectMetadata objectMetadata = newObjectMetadata(length);
-    if (headers != null) {
-      objectMetadata.setUserMetadata(headers);
-    }
     return getRequestFactory().newPutObjectRequest(
         destKey,
         objectMetadata,
+        options,
         inputStream);
   }
 
   /**
    * Create a {@link PutObjectRequest} request to upload a file.
    * @param dest key to PUT to.
    * @param sourceFile source file
-   * @param headers optional map of custom headers.
+   * @param options options for the request
    * @return the request
    */
   @Retries.OnceRaw
   public PutObjectRequest createPutObjectRequest(
       String dest,
       File sourceFile,
-      @Nullable final Map<String, String> headers) {
+      final PutObjectOptions options) {
     Preconditions.checkState(sourceFile.length() < Integer.MAX_VALUE,
         "File length is too big for a single PUT upload");
     activateAuditSpan();
     final ObjectMetadata objectMetadata =
         newObjectMetadata((int) sourceFile.length());
-    if (headers != null) {
-      objectMetadata.setUserMetadata(headers);
-    }
-    return getRequestFactory().
+
+    PutObjectRequest putObjectRequest = getRequestFactory().
         newPutObjectRequest(dest,
             objectMetadata,
+            options,
             sourceFile);
+    return putObjectRequest;
   }
 
   /**
@@ -319,7 +316,7 @@ public String initiateMultiPartUpload(
           () -> {
             final InitiateMultipartUploadRequest initiateMPURequest =
                 getRequestFactory().newMultipartUploadRequest(
-                    destKey, options.getHeaders());
+                    destKey, options);
             return owner.initiateMultipartUpload(initiateMPURequest)
                 .getUploadId();
           });

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

@@ -25,7 +25,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import com.amazonaws.services.s3.model.CompleteMultipartUploadResult;
@@ -79,25 +78,25 @@ <T> T retry(String action,
    * @param destKey destination key
    * @param inputStream source data.
    * @param length size, if known. Use -1 for not known
-   * @param headers optional map of custom headers.
+   * @param options
    * @return the request
    */
   PutObjectRequest createPutObjectRequest(String destKey,
       InputStream inputStream,
       long length,
-      @Nullable Map<String, String> headers);
+      @Nullable PutObjectOptions options);
 
   /**
    * Create a {@link PutObjectRequest} request to upload a file.
    * @param dest key to PUT to.
    * @param sourceFile source file
-   * @param headers optional map of custom headers.
+   * @param options
    * @return the request
    */
   PutObjectRequest createPutObjectRequest(
       String dest,
       File sourceFile,
-      @Nullable Map<String, String> headers);
+      @Nullable PutObjectOptions options);
 
   /**
    * Callback on a successful write.

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

@@ -22,7 +22,6 @@
 import java.io.File;
 import java.io.InputStream;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 
 import com.amazonaws.services.s3.model.AbortMultipartUploadRequest;
@@ -51,6 +50,7 @@
 import org.apache.hadoop.fs.PathIOException;
 import org.apache.hadoop.fs.s3a.S3AEncryptionMethods;
 import org.apache.hadoop.fs.s3a.auth.delegation.EncryptionSecrets;
+import org.apache.hadoop.fs.s3a.impl.PutObjectOptions;
 
 /**
  * Factory for S3 objects.
@@ -142,23 +142,26 @@ CopyObjectRequest newCopyObjectRequest(String srcKey,
    * Adds the ACL and metadata
    * @param key key of object
    * @param metadata metadata header
+   * @param options
    * @param srcfile source file
    * @return the request
    */
   PutObjectRequest newPutObjectRequest(String key,
-      ObjectMetadata metadata, File srcfile);
+      ObjectMetadata metadata, final PutObjectOptions options, File srcfile);
 
   /**
    * Create a {@link PutObjectRequest} request.
    * The metadata is assumed to have been configured with the size of the
    * operation.
    * @param key key of object
    * @param metadata metadata header
+   * @param options options for the request
    * @param inputStream source data.
    * @return the request
    */
   PutObjectRequest newPutObjectRequest(String key,
       ObjectMetadata metadata,
+      PutObjectOptions options,
       InputStream inputStream);
 
   /**
@@ -196,7 +199,7 @@ AbortMultipartUploadRequest newAbortMultipartUploadRequest(
    */
   InitiateMultipartUploadRequest newMultipartUploadRequest(
       String destKey,
-      @Nullable Map<String, String> headers);
+      @Nullable PutObjectOptions options);
 
   /**
    * Complete a multipart upload.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/magic/MagicCommitTracker.java

@@ -146,7 +146,7 @@ public boolean aboutToComplete(String uploadId,
         originalDestKey,
         new ByteArrayInputStream(EMPTY),
         0,
-        headers);
+        new PutObjectOptions(true, null, headers));
     upload(originalDestPut);
 
     // build the commit summary

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

@@ -31,6 +31,11 @@ public final class PutObjectOptions {
    */
   private final boolean keepMarkers;
 
+  /**
+   * Storage class, if not null.
+   */
+  private final String storageClass;
+
   /**
    * Headers; may be null.
    */
@@ -39,12 +44,15 @@ public final class PutObjectOptions {
   /**
    * Constructor.
    * @param keepMarkers Can the PUT operation skip marker deletion?
+   * @param storageClass Storage class, if not null.
    * @param headers Headers; may be null.
    */
   public PutObjectOptions(
       final boolean keepMarkers,
+      @Nullable final String storageClass,
       @Nullable final Map<String, String> headers) {
     this.keepMarkers = keepMarkers;
+    this.storageClass = storageClass;
     this.headers = headers;
   }
 
@@ -68,11 +76,14 @@ public Map<String, String> getHeaders() {
   public String toString() {
     return "PutObjectOptions{" +
         "keepMarkers=" + keepMarkers +
+        ", storageClass='" + storageClass + '\'' +
         '}';
   }
 
-  private static final PutObjectOptions KEEP_DIRS = new PutObjectOptions(true, null);
-  private static final PutObjectOptions DELETE_DIRS = new PutObjectOptions(false,  null);
+  private static final PutObjectOptions KEEP_DIRS = new PutObjectOptions(true,
+      null, null);
+  private static final PutObjectOptions DELETE_DIRS = new PutObjectOptions(false,
+      null, null);
 
   /**
    * Get the options to keep directories.
@@ -89,4 +100,5 @@ public static PutObjectOptions keepingDirs() {
   public static PutObjectOptions deletingDirs() {
     return DELETE_DIRS;
   }
+
 }

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

@@ -25,7 +25,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-
 import javax.annotation.Nullable;
 
 import com.amazonaws.AmazonWebServiceRequest;
@@ -365,15 +364,19 @@ protected void copyEncryptionParameters(
    * Adds the ACL, storage class and metadata
    * @param key key of object
    * @param metadata metadata header
+   * @param options options for the request, including headers
    * @param srcfile source file
    * @return the request
    */
   @Override
   public PutObjectRequest newPutObjectRequest(String key,
-      ObjectMetadata metadata, File srcfile) {
+      ObjectMetadata metadata,
+      final PutObjectOptions options,
+      File srcfile) {
     Preconditions.checkNotNull(srcfile);
     PutObjectRequest putObjectRequest = new PutObjectRequest(getBucket(), key,
         srcfile);
+    maybeSetMetadata(options, metadata);
     setOptionalPutRequestParameters(putObjectRequest);
     putObjectRequest.setCannedAcl(cannedACL);
     if (storageClass != null) {
@@ -389,15 +392,18 @@ public PutObjectRequest newPutObjectRequest(String key,
    * operation.
    * @param key key of object
    * @param metadata metadata header
+   * @param options options for the request
    * @param inputStream source data.
    * @return the request
    */
   @Override
   public PutObjectRequest newPutObjectRequest(String key,
       ObjectMetadata metadata,
+      @Nullable final PutObjectOptions options,
       InputStream inputStream) {
     Preconditions.checkNotNull(inputStream);
     Preconditions.checkArgument(isNotEmpty(key), "Null/empty key");
+    maybeSetMetadata(options, metadata);
     PutObjectRequest putObjectRequest = new PutObjectRequest(getBucket(), key,
         inputStream, metadata);
     setOptionalPutRequestParameters(putObjectRequest);
@@ -421,7 +427,7 @@ public int read() throws IOException {
     final ObjectMetadata md = createObjectMetadata(0L, true);
     md.setContentType(HeaderProcessing.CONTENT_TYPE_X_DIRECTORY);
     PutObjectRequest putObjectRequest =
-        newPutObjectRequest(key, md, im);
+        newPutObjectRequest(key, md, null, im);
     return putObjectRequest;
   }
 
@@ -447,12 +453,10 @@ public AbortMultipartUploadRequest newAbortMultipartUploadRequest(
 
   @Override
   public InitiateMultipartUploadRequest newMultipartUploadRequest(
-      String destKey,
-      @Nullable final Map<String, String> headers) {
+      final String destKey,
+      @Nullable final PutObjectOptions options) {
     final ObjectMetadata objectMetadata = newObjectMetadata(-1);
-    if (headers != null) {
-      objectMetadata.setUserMetadata(headers);
-    }
+    maybeSetMetadata(options, objectMetadata);
     final InitiateMultipartUploadRequest initiateMPURequest =
         new InitiateMultipartUploadRequest(getBucket(),
             destKey,
@@ -609,6 +613,23 @@ public void setEncryptionSecrets(final EncryptionSecrets secrets) {
     encryptionSecrets = secrets;
   }
 
+  /**
+   * Set the metadata from the options if the options are not
+   * null and the metadata contains headers.
+   * @param options options for the request
+   * @param objectMetadata metadata to patch
+   */
+  private void maybeSetMetadata(
+      @Nullable PutObjectOptions options,
+      final ObjectMetadata objectMetadata) {
+    if (options != null) {
+      Map<String, String> headers = options.getHeaders();
+      if (headers != null) {
+        objectMetadata.setUserMetadata(headers);
+      }
+    }
+  }
+
   /**
    * Create a builder.
    * @return new builder.

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestRequestFactory.java

@@ -90,7 +90,7 @@ public void testRequestFactoryWithCannedACL() throws Throwable {
     ObjectMetadata md = factory.newObjectMetadata(128);
     Assertions.assertThat(
             factory.newPutObjectRequest(path, md,
-                    new ByteArrayInputStream(new byte[0]))
+                    null, new ByteArrayInputStream(new byte[0]))
                 .getCannedAcl())
         .describedAs("ACL of PUT")
         .isEqualTo(acl);
@@ -176,9 +176,9 @@ private void createFactoryObjects(RequestFactory factory) {
     a(factory.newMultipartUploadRequest(path, null));
     File srcfile = new File("/tmp/a");
     a(factory.newPutObjectRequest(path,
-        factory.newObjectMetadata(-1), srcfile));
+        factory.newObjectMetadata(-1), null, srcfile));
     ByteArrayInputStream stream = new ByteArrayInputStream(new byte[0]);
-    a(factory.newPutObjectRequest(path, md, stream));
+    a(factory.newPutObjectRequest(path, md, null, stream));
     a(factory.newSelectRequest(path));
   }
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3ADirectoryPerformance.java

@@ -258,7 +258,7 @@ public void testMultiPagesListingPerformanceAndCorrectness()
         ObjectMetadata om = fs.newObjectMetadata(0L);
         PutObjectRequest put = requestFactory
             .newPutObjectRequest(fs.pathToKey(file), om,
-                new FailingInputStream());
+                null, new FailingInputStream());
         futures.add(submit(executorService, () ->
             writeOperationHelper.putObject(put, PutObjectOptions.keepingDirs())));
       }