Merge branch 'apache:trunk' into YARN-11267-V2

Author: slfan1989

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

@@ -1781,6 +1781,15 @@ private Constants() {
    */
   public static final boolean CHECKSUM_VALIDATION_DEFAULT = false;
 
+  /**
+   * Indicates the algorithm used to create the checksum for the object
+   * to be uploaded to S3. Unset by default. It supports the following values:
+   * 'CRC32', 'CRC32C', 'SHA1', and 'SHA256'
+   * value:{@value}
+   */
+  public static final String CHECKSUM_ALGORITHM =
+      "fs.s3a.create.checksum.algorithm";
+
   /**
    * Are extensions classes, such as {@code fs.s3a.aws.credentials.provider},
    * going to be loaded from the same classloader that loaded

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

@@ -1064,6 +1064,10 @@ private void uploadBlockAsync(final S3ADataBlocks.DataBlock block,
               return CompletedPart.builder()
                   .eTag(response.eTag())
                   .partNumber(currentPartNumber)
+                  .checksumCRC32(response.checksumCRC32())
+                  .checksumCRC32C(response.checksumCRC32C())
+                  .checksumSHA1(response.checksumSHA1())
+                  .checksumSHA256(response.checksumSHA256())
                   .build();
             } catch (Exception e) {
               final IOException ex = e instanceof IOException

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

@@ -117,6 +117,7 @@
 import org.apache.hadoop.fs.s3a.impl.BulkDeleteOperationCallbacksImpl;
 import org.apache.hadoop.fs.s3a.impl.CSES3AFileSystemOperations;
 import org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy;
+import org.apache.hadoop.fs.s3a.impl.ChecksumSupport;
 import org.apache.hadoop.fs.s3a.impl.ClientManager;
 import org.apache.hadoop.fs.s3a.impl.ClientManagerImpl;
 import org.apache.hadoop.fs.s3a.impl.ConfigurationHelper;
@@ -1327,6 +1328,7 @@ protected RequestFactory createRequestFactory() {
         .withStorageClass(storageClass)
         .withMultipartUploadEnabled(isMultipartUploadEnabled)
         .withPartUploadTimeout(partUploadTimeout)
+        .withChecksumAlgorithm(ChecksumSupport.getChecksumAlgorithm(getConf()))
         .build();
   }
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/files/SinglePendingCommit.java

@@ -71,7 +71,7 @@
 @InterfaceAudience.Private
 @InterfaceStability.Unstable
 public class SinglePendingCommit extends PersistentCommitData<SinglePendingCommit>
-    implements Iterable<String> {
+    implements Iterable<UploadEtag> {
 
   /**
    * Serialization ID: {@value}.
@@ -118,7 +118,7 @@ public class SinglePendingCommit extends PersistentCommitData<SinglePendingCommi
   private String text = "";
 
   /** Ordered list of etags. */
-  private List<String> etags;
+  private List<UploadEtag> etags;
 
   /**
    * Any custom extra data committer subclasses may choose to add.
@@ -222,7 +222,7 @@ public void bindCommitData(List<CompletedPart> parts) throws ValidationFailure {
     for (CompletedPart part : parts) {
       verify(part.partNumber() == counter,
           "Expected part number %s but got %s", counter, part.partNumber());
-      etags.add(part.eTag());
+      etags.add(UploadEtag.fromCompletedPart(part));
       counter++;
     }
   }
@@ -237,9 +237,10 @@ public void validate() throws ValidationFailure {
     verify(length >= 0, "Invalid length: " + length);
     destinationPath();
     verify(etags != null, "No etag list");
-    validateCollectionClass(etags, String.class);
-    for (String etag : etags) {
-      verify(StringUtils.isNotEmpty(etag), "Empty etag");
+    validateCollectionClass(etags, UploadEtag.class);
+    for (UploadEtag etag : etags) {
+      verify(etag != null && StringUtils.isNotEmpty(etag.getEtag()),
+          "Empty etag");
     }
     if (extraData != null) {
       validateCollectionClass(extraData.keySet(), String.class);
@@ -313,7 +314,7 @@ public int getPartCount() {
    * @return an iterator.
    */
   @Override
-  public Iterator<String> iterator() {
+  public Iterator<UploadEtag> iterator() {
     return etags.iterator();
   }
 
@@ -442,11 +443,11 @@ public void setText(String text) {
   }
 
   /** @return ordered list of etags. */
-  public List<String> getEtags() {
+  public List<UploadEtag> getEtags() {
     return etags;
   }
 
-  public void setEtags(List<String> etags) {
+  public void setEtags(List<UploadEtag> etags) {
     this.etags = etags;
   }
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/files/UploadEtag.java

@@ -0,0 +1,136 @@
+/*
+ * 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.commit.files;
+
+import java.io.Serializable;
+import java.util.StringJoiner;
+
+import software.amazon.awssdk.services.s3.model.ChecksumAlgorithm;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+
+/**
+ * Stores ETag and checksum values from {@link  CompletedPart} responses from S3.
+ * These values need to be stored to be later passed to the
+ * {@link software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest
+ * CompleteMultipartUploadRequest}
+ */
+public class UploadEtag implements Serializable {
+
+  /**
+   * Serialization ID: {@value}.
+   */
+  private static final long serialVersionUID = 1L;
+
+  private String etag;
+  private String checksumAlgorithm;
+  private String checksum;
+
+  public UploadEtag() {
+  }
+
+  public UploadEtag(String etag, String checksumAlgorithm, String checksum) {
+    this.etag = etag;
+    this.checksumAlgorithm = checksumAlgorithm;
+    this.checksum = checksum;
+  }
+
+  public String getEtag() {
+    return etag;
+  }
+
+  public void setEtag(String etag) {
+    this.etag = etag;
+  }
+
+  public String getChecksumAlgorithm() {
+    return checksumAlgorithm;
+  }
+
+  public void setChecksumAlgorithm(String checksumAlgorithm) {
+    this.checksumAlgorithm = checksumAlgorithm;
+  }
+
+  public String getChecksum() {
+    return checksum;
+  }
+
+  public void setChecksum(String checksum) {
+    this.checksum = checksum;
+  }
+
+  public static UploadEtag fromCompletedPart(CompletedPart completedPart) {
+    UploadEtag uploadEtag = new UploadEtag();
+    uploadEtag.setEtag(completedPart.eTag());
+    if (completedPart.checksumCRC32() != null) {
+      uploadEtag.setChecksumAlgorithm(ChecksumAlgorithm.CRC32.toString());
+      uploadEtag.setChecksum(completedPart.checksumCRC32());
+    }
+    if (completedPart.checksumCRC32C() != null) {
+      uploadEtag.setChecksumAlgorithm(ChecksumAlgorithm.CRC32_C.toString());
+      uploadEtag.setChecksum(completedPart.checksumCRC32C());
+    }
+    if (completedPart.checksumSHA1() != null) {
+      uploadEtag.setChecksumAlgorithm(ChecksumAlgorithm.SHA1.toString());
+      uploadEtag.setChecksum(completedPart.checksumSHA1());
+    }
+    if (completedPart.checksumSHA256() != null) {
+      uploadEtag.setChecksumAlgorithm(ChecksumAlgorithm.SHA256.toString());
+      uploadEtag.setChecksum(completedPart.checksumSHA256());
+    }
+    return uploadEtag;
+  }
+
+  public static CompletedPart toCompletedPart(UploadEtag uploadEtag, int partNumber) {
+    final CompletedPart.Builder builder = CompletedPart.builder()
+        .partNumber(partNumber)
+        .eTag(uploadEtag.etag);
+    if (uploadEtag.checksumAlgorithm == null) {
+      return builder.build();
+    }
+    final ChecksumAlgorithm checksumAlgorithm = ChecksumAlgorithm.fromValue(
+        uploadEtag.checksumAlgorithm);
+    switch (checksumAlgorithm) {
+    case CRC32:
+      builder.checksumCRC32(uploadEtag.checksum);
+      break;
+    case CRC32_C:
+      builder.checksumCRC32C(uploadEtag.checksum);
+      break;
+    case SHA1:
+      builder.checksumSHA1(uploadEtag.checksum);
+      break;
+    case SHA256:
+      builder.checksumSHA256(uploadEtag.checksum);
+      break;
+    default:
+      // do nothing
+    }
+    return builder.build();
+  }
+
+  @Override
+  public String toString() {
+    return new StringJoiner(", ", UploadEtag.class.getSimpleName() + "[", "]")
+        .add("serialVersionUID='" + serialVersionUID + "'")
+        .add("etag='" + etag + "'")
+        .add("checksumAlgorithm='" + checksumAlgorithm + "'")
+        .add("checksum='" + checksum + "'")
+        .toString();
+  }
+}

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

@@ -53,6 +53,7 @@
 import org.apache.hadoop.fs.s3a.WriteOperations;
 import org.apache.hadoop.fs.s3a.commit.CommitConstants;
 import org.apache.hadoop.fs.s3a.commit.PathCommitException;
+import org.apache.hadoop.fs.s3a.commit.files.UploadEtag;
 import org.apache.hadoop.fs.s3a.commit.files.PendingSet;
 import org.apache.hadoop.fs.s3a.commit.files.SinglePendingCommit;
 import org.apache.hadoop.fs.s3a.commit.files.SuccessData;
@@ -165,9 +166,9 @@ public CommitOperations(S3AFileSystem fs,
    * @param tagIds list of tags
    * @return same list, now in numbered tuples
    */
-  public static List<CompletedPart> toPartEtags(List<String> tagIds) {
+  public static List<CompletedPart> toPartEtags(List<UploadEtag> tagIds) {
     return IntStream.range(0, tagIds.size())
-        .mapToObj(i -> CompletedPart.builder().partNumber(i + 1).eTag(tagIds.get(i)).build())
+        .mapToObj(i -> UploadEtag.toCompletedPart(tagIds.get(i), i + 1))
         .collect(Collectors.toList());
   }
 
@@ -655,6 +656,10 @@ private List<CompletedPart> uploadFileData(
       parts.add(CompletedPart.builder()
           .partNumber(partNumber)
           .eTag(response.eTag())
+          .checksumCRC32(response.checksumCRC32())
+          .checksumCRC32C(response.checksumCRC32C())
+          .checksumSHA1(response.checksumSHA1())
+          .checksumSHA256(response.checksumSHA256())
           .build());
     }
     return parts;

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

@@ -0,0 +1,71 @@
+/*
+ * 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.impl;
+
+import java.util.Set;
+
+import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableSet;
+import software.amazon.awssdk.services.s3.model.ChecksumAlgorithm;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.util.ConfigurationHelper;
+
+import static org.apache.hadoop.fs.s3a.Constants.CHECKSUM_ALGORITHM;
+
+/**
+ * Utility class to support operations on S3 object checksum.
+ */
+public final class ChecksumSupport {
+
+  private ChecksumSupport() {
+  }
+
+  /**
+   * Checksum algorithms that are supported by S3A.
+   */
+  private static final Set<ChecksumAlgorithm> SUPPORTED_CHECKSUM_ALGORITHMS = ImmutableSet.of(
+      ChecksumAlgorithm.CRC32,
+      ChecksumAlgorithm.CRC32_C,
+      ChecksumAlgorithm.SHA1,
+      ChecksumAlgorithm.SHA256);
+
+  /**
+   * Get the checksum algorithm to be used for data integrity check of the objects in S3.
+   * This operation includes validating if the provided value is a supported checksum algorithm.
+   * @param conf configuration to scan
+   * @return the checksum algorithm to be passed on S3 requests
+   * @throws IllegalArgumentException if the checksum algorithm is not known or not supported
+   */
+  public static ChecksumAlgorithm getChecksumAlgorithm(Configuration conf) {
+    return ConfigurationHelper.resolveEnum(conf,
+        CHECKSUM_ALGORITHM,
+        ChecksumAlgorithm.class,
+        configValue -> {
+          if (StringUtils.isBlank(configValue)) {
+            return null;
+          }
+          if (ChecksumAlgorithm.CRC32_C.toString().equalsIgnoreCase(configValue)) {
+            // In case the configuration value is CRC32C, without underscore.
+            return ChecksumAlgorithm.CRC32_C;
+          }
+          throw new IllegalArgumentException("Checksum algorithm is not supported: " + configValue);
+        });
+  }
+}