HADOOP-17851 Support user specified content encoding for S3A

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

@@ -410,6 +410,10 @@ private Constants() {
   public static final String CANNED_ACL = "fs.s3a.acl.default";
   public static final String DEFAULT_CANNED_ACL = "";
 
+  // gzip, deflate, compress, br, etc.
+  public static final String CONTENT_ENCODING = "fs.s3a.contentEncoding";
+  public static final String DEFAULT_CONTENT_ENCODING = null;
+
   // should we try to purge old multipart uploads when starting up
   public static final String PURGE_EXISTING_MULTIPART =
       "fs.s3a.multipart.purge";

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

@@ -925,12 +925,16 @@ protected RequestFactory createRequestFactory() {
     // request factory.
     initCannedAcls(getConf());
 
+    // Any encoding type
+    String contentEncoding = getConf().get(CONTENT_ENCODING, DEFAULT_CONTENT_ENCODING);
+
     return RequestFactoryImpl.builder()
         .withBucket(requireNonNull(bucket))
         .withCannedACL(getCannedACL())
         .withEncryptionSecrets(requireNonNull(encryptionSecrets))
         .withMultipartPartCountLimit(partCountLimit)
         .withRequestPreparer(getAuditManager()::requestCreated)
+        .withContentEncoding(contentEncoding)
         .build();
   }
 

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

@@ -100,6 +100,12 @@ public interface RequestFactory {
    */
   S3AEncryptionMethods getServerSideEncryptionAlgorithm();
 
+  /**
+   * Get the content encoding (e.g. gzip) or return null if none.
+   * @return content encoding
+   */
+  public String getContentEncoding();
+
   /**
    * Create a new object metadata instance.
    * Any standard metadata headers are added here, for example:

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

@@ -118,6 +118,11 @@ public class RequestFactoryImpl implements RequestFactory {
    */
   private final PrepareRequest requestPreparer;
 
+  /**
+   * Content encoding (null for none).
+   */
+  private final String contentEncoding;
+
   /**
    * Constructor.
    * @param builder builder with all the configuration.
@@ -130,6 +135,7 @@ protected RequestFactoryImpl(
     this.multipartPartCountLimit = builder.multipartPartCountLimit;
     this.requesterPays = builder.requesterPays;
     this.requestPreparer = builder.requestPreparer;
+    this.contentEncoding = builder.contentEncoding;
   }
 
   /**
@@ -193,6 +199,15 @@ public S3AEncryptionMethods getServerSideEncryptionAlgorithm() {
     return encryptionSecrets.getEncryptionMethod();
   }
 
+  /**
+   * Get the content encoding (e.g. gzip) or return null if none.
+   * @return content encoding
+   */
+  @Override
+  public String getContentEncoding() {
+    return contentEncoding;
+  }
+
   /**
    * Sets server side encryption parameters to the part upload
    * request when encryption is enabled.
@@ -238,11 +253,16 @@ protected void setOptionalPutRequestParameters(PutObjectRequest request) {
    * @param metadata to update.
    */
   protected void setOptionalObjectMetadata(ObjectMetadata metadata) {
+    final String contentEncoding =
+      getContentEncoding();
     final S3AEncryptionMethods algorithm
         = getServerSideEncryptionAlgorithm();
     if (S3AEncryptionMethods.SSE_S3 == algorithm) {
       metadata.setSSEAlgorithm(algorithm.getMethod());
     }
+    if (contentEncoding != null) {
+      metadata.setContentEncoding(contentEncoding);
+    }
   }
 
   /**
@@ -586,6 +606,9 @@ public static final class RequestFactoryBuilder {
     /** Requester Pays flag. */
     private boolean requesterPays = false;
 
+    /** Content Encoding. */
+    private String contentEncoding = null;
+
     /**
      * Multipart limit.
      */
@@ -607,6 +630,16 @@ public RequestFactory build() {
       return new RequestFactoryImpl(this);
     }
 
+    /**
+     * Content encoding.
+     * @param value new value
+     * @return the builder
+     */
+    public RequestFactoryBuilder withContentEncoding(final String value) {
+      contentEncoding = value;
+      return this;
+    }
+
     /**
      * Target bucket.
      * @param value new value

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

@@ -0,0 +1,82 @@
+/*
+ * 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 com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.ObjectMetadata;
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.apache.hadoop.fs.s3a.audit.S3AAuditConstants;
+import org.apache.hadoop.fs.s3a.impl.StoreContext;
+
+import static org.apache.hadoop.fs.s3a.Constants.CONTENT_ENCODING;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
+
+/**
+ * Tests of content encoding object meta data.
+ */
+public class ITestS3AContentEncoding extends AbstractS3ATestBase {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(ITestS3ACannedACLs.class);
+
+  @Override
+  protected Configuration createConfiguration() {
+    Configuration conf = super.createConfiguration();
+    conf.set(CONTENT_ENCODING, "gzip");
+
+    return conf;
+  }
+
+  @Test
+  public void testCreatedObjectsHaveEncoding() throws Throwable {
+    S3AFileSystem fs = getFileSystem();
+    Path dir = methodPath();
+    fs.mkdirs(dir);
+    Path path = new Path(dir, "1");
+    ContractTestUtils.touch(fs, path);
+    assertObjectHasEncoding(path);
+  }
+
+  /**
+   * Assert that a given object has gzip encoding specified
+   * @param path path
+   */
+  private void assertObjectHasEncoding(Path path) {
+    S3AFileSystem fs = getFileSystem();
+
+    StoreContext storeContext = fs.createStoreContext();
+    AmazonS3 s3 = fs.getAmazonS3ClientForTesting("encoding");
+    String key = storeContext.pathToKey(path);
+    ObjectMetadata meta = s3.getObjectMetadata(storeContext.getBucket(),
+        key);
+    String encoding = meta.getContentEncoding();
+    Assertions.assertThat(encoding)
+        .describedAs("Encoding of object %s is gzip", path)
+        .isEqualTo("gzip");
+  }
+}

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

@@ -39,8 +39,9 @@ protected Configuration createConfiguration() {
     // get the KMS key for this test.
     Configuration c = new Configuration();
     String kmsKey = c.get(SERVER_SIDE_ENCRYPTION_KEY);
-    if (StringUtils.isBlank(kmsKey) || !c.get(SERVER_SIDE_ENCRYPTION_ALGORITHM)
-        .equals(S3AEncryptionMethods.CSE_KMS.name())) {
+    String encryptionAlgorithm = c.get(SERVER_SIDE_ENCRYPTION_ALGORITHM);
+    if (kmsKey == null || StringUtils.isBlank(kmsKey) || encryptionAlgorithm == null ||
+        !encryptionAlgorithm.equals(S3AEncryptionMethods.CSE_KMS.name())) {
       skip(SERVER_SIDE_ENCRYPTION_KEY + " is not set for " +
           SSE_KMS.getMethod() + " or CSE-KMS algorithm is used instead of "
           + "SSE-KMS");

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

@@ -45,8 +45,9 @@ public class ITestS3AHugeFilesEncryption extends AbstractSTestS3AHugeFiles {
   public void setup() throws Exception {
     Configuration c = new Configuration();
     String kmsKey = c.get(SERVER_SIDE_ENCRYPTION_KEY);
-    if (StringUtils.isBlank(kmsKey) || !c.get(SERVER_SIDE_ENCRYPTION_ALGORITHM)
-        .equals(S3AEncryptionMethods.CSE_KMS.name())) {
+    String encryptionAlgorithm = c.get(SERVER_SIDE_ENCRYPTION_ALGORITHM);
+    if (kmsKey == null || StringUtils.isBlank(kmsKey) || encryptionAlgorithm == null ||
+        !encryptionAlgorithm.equals(S3AEncryptionMethods.CSE_KMS.name())) {
       skip(SERVER_SIDE_ENCRYPTION_KEY + " is not set for " +
           SSE_KMS.getMethod() + " or CSE-KMS algorithm is used instead of "
           + "SSE-KMS");