Production Code changes to support Checksum

Author: Anuj Modi

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsConfiguration.java

@@ -337,6 +337,10 @@ public class AbfsConfiguration{
           FS_AZURE_ABFS_RENAME_RESILIENCE, DefaultValue = DEFAULT_ENABLE_ABFS_RENAME_RESILIENCE)
   private boolean renameResilience;
 
+  @BooleanConfigurationValidatorAnnotation(ConfigurationKey =
+      FS_AZURE_ABFS_ENABLE_CHECKSUM, DefaultValue = DEFAULT_ENABLE_ABFS_CHECKSUM)
+  private boolean isChecksumEnabled;
+
   public AbfsConfiguration(final Configuration rawConfig, String accountName)
       throws IllegalAccessException, InvalidConfigurationValueException, IOException {
     this.rawConfig = ProviderUtils.excludeIncompatibleCredentialProviders(
@@ -1150,4 +1154,12 @@ public boolean getRenameResilience() {
   void setRenameResilience(boolean actualResilience) {
     renameResilience = actualResilience;
   }
+
+  public boolean getIsChecksumEnabled() {
+    return isChecksumEnabled;
+  }
+
+  void setIsChecksumEnabled(boolean isChecksumEnabled) {
+    this.isChecksumEnabled = isChecksumEnabled;
+  }
 }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/ConfigurationKeys.java

@@ -241,6 +241,9 @@ public final class ConfigurationKeys {
   /** Add extra resilience to rename failures, at the expense of performance. */
   public static final String FS_AZURE_ABFS_RENAME_RESILIENCE = "fs.azure.enable.rename.resilience";
 
+  /** Add extra integrity checks on data read and written using Md5 Hash Validation*/
+  public static final String FS_AZURE_ABFS_ENABLE_CHECKSUM = "fs.azure.enable.checksum";
+
   public static String accountProperty(String property, String account) {
     return property + "." + account;
   }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/FileSystemConfigurations.java

@@ -119,6 +119,7 @@ public final class FileSystemConfigurations {
   public static final int STREAM_ID_LEN = 12;
   public static final boolean DEFAULT_ENABLE_ABFS_LIST_ITERATOR = true;
   public static final boolean DEFAULT_ENABLE_ABFS_RENAME_RESILIENCE = true;
+  public static final boolean DEFAULT_ENABLE_ABFS_CHECKSUM = false;
 
   /**
    * Limit of queued block upload operations before writes

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/HttpHeaderConfigurations.java

@@ -71,6 +71,7 @@ public final class HttpHeaderConfigurations {
   public static final String X_MS_PROPOSED_LEASE_ID = "x-ms-proposed-lease-id";
   public static final String X_MS_LEASE_BREAK_PERIOD = "x-ms-lease-break-period";
   public static final String EXPECT = "Expect";
+  public static final String X_MS_RANGE_GET_CONTENT_MD5 = "x-ms-range-get-content-md5";
 
   private HttpHeaderConfigurations() {}
 }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java

@@ -49,6 +49,7 @@
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.MoreExecutors;
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
 
+import com.sun.tools.javac.util.Convert;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -75,6 +76,7 @@
 import static org.apache.hadoop.fs.azurebfs.AzureBlobFileSystemStore.extractEtagHeader;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.*;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DEFAULT_DELETE_CONSIDERED_IDEMPOTENT;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_MB;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.SERVER_SIDE_ENCRYPTION_ALGORITHM;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.HTTPS_SCHEME;
 import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.*;
@@ -761,6 +763,8 @@ public AbfsRestOperation append(final String path, final byte[] buffer,
       requestHeaders.add(new AbfsHttpHeader(USER_AGENT, userAgentRetry));
     }
 
+    addCheckSumHeaderForWrite(requestHeaders, buffer);
+
     // AbfsInputStream/AbfsOutputStream reuse SAS tokens for better performance
     String sasTokenForReuse = appendSASTokenToQuery(path, SASTokenProvider.WRITE_OPERATION,
         abfsUriQueryBuilder, cachedSasToken);
@@ -978,9 +982,12 @@ public AbfsRestOperation read(final String path, final long position, final byte
       TracingContext tracingContext) throws AzureBlobFileSystemException {
     final List<AbfsHttpHeader> requestHeaders = createDefaultHeaders();
     addCustomerProvidedKeyHeaders(requestHeaders);
-    requestHeaders.add(new AbfsHttpHeader(RANGE,
-            String.format("bytes=%d-%d", position, position + bufferLength - 1)));
+
+    AbfsHttpHeader rangeHeader = new AbfsHttpHeader(RANGE,
+        String.format("bytes=%d-%d", position, position + bufferLength - 1));
+    requestHeaders.add(rangeHeader);
     requestHeaders.add(new AbfsHttpHeader(IF_MATCH, eTag));
+    addCheckSumHeaderForRead(requestHeaders, bufferLength, rangeHeader);
 
     final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
     // AbfsInputStream/AbfsOutputStream reuse SAS tokens for better performance
@@ -999,6 +1006,8 @@ public AbfsRestOperation read(final String path, final long position, final byte
             bufferLength, sasTokenForReuse);
     op.execute(tracingContext);
 
+    verifyCheckSumForRead(buffer, op.getResult());
+
     return op;
   }
 
@@ -1412,6 +1421,54 @@ private void appendIfNotEmpty(StringBuilder sb, String regEx,
     }
   }
 
+  private void addCheckSumHeaderForRead(List<AbfsHttpHeader> requestHeaders,
+      final int bufferLength, final AbfsHttpHeader rangeHeader) {
+    if(getAbfsConfiguration().getIsChecksumEnabled() &&
+        requestHeaders.contains(rangeHeader) && bufferLength <= 4 * ONE_MB) {
+       requestHeaders.add(new AbfsHttpHeader(X_MS_RANGE_GET_CONTENT_MD5, TRUE));
+    }
+  }
+
+  private void addCheckSumHeaderForWrite(List<AbfsHttpHeader> requestHeaders,
+      final byte[] buffer) {
+    if(getAbfsConfiguration().getIsChecksumEnabled()) {
+      try {
+        MessageDigest md5Digest = MessageDigest.getInstance("MD5");
+        byte[] md5Bytes = md5Digest.digest(buffer);
+        String md5Hash = Base64.getEncoder().encodeToString(md5Bytes);
+        requestHeaders.add(new AbfsHttpHeader(CONTENT_MD5, md5Hash));
+      } catch (NoSuchAlgorithmException e) {
+        e.printStackTrace();
+      }
+    }
+  }
+
+  private void verifyCheckSumForRead(final byte[] buffer, final AbfsHttpOperation result)
+      throws AbfsRestOperationException{
+    if(getAbfsConfiguration().getIsChecksumEnabled()) {
+      // Number of bytes returned by server could be less than or equal to what
+      // caller requests. In case it is less, extra bytes will be initialized to 0
+      // Server returned MD5 Hash will be computed on what server returned.
+      // We need to get exact data that server returned and compute its md5 hash
+      // Computed hash should be equal to what server returned
+      int numberOfBytesRead = (int)result.getBytesReceived();
+      byte[] dataRead = new byte[numberOfBytesRead];
+      System.arraycopy(buffer, 0, dataRead, 0, numberOfBytesRead);
+
+      try {
+        MessageDigest md5Digest = MessageDigest.getInstance("MD5");
+        byte[] md5Bytes = md5Digest.digest(dataRead);
+        String md5HashComputed = Base64.getEncoder().encodeToString(md5Bytes);
+        String md5HashActual = result.getResponseHeader(CONTENT_MD5);
+        if (!md5HashComputed.equals(md5HashActual)) {
+          throw new AbfsRestOperationException(-1, "-1", "Checksum Check Failed", new IOException());
+        }
+      } catch (NoSuchAlgorithmException e) {
+        e.printStackTrace();
+      }
+    }
+  }
+
   @VisibleForTesting
   URL getBaseUrl() {
     return baseUrl;

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemChecksum.java

@@ -0,0 +1,62 @@
+/**
+ * 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.azurebfs;
+
+import java.nio.charset.StandardCharsets;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
+
+/**
+ * Test For Verifying Checksum Related Operations
+ */
+public class ITestAzureBlobFileSystemChecksum extends AbstractAbfsIntegrationTest {
+
+  public ITestAzureBlobFileSystemChecksum() throws Exception {
+    super();
+  }
+
+  @Test
+  public void testWriteReadWithChecksum() throws Exception {
+    AzureBlobFileSystem fs = getFileSystem();
+    AbfsConfiguration conf = fs.getAbfsStore().getAbfsConfiguration();
+    // Enable checksum validations for Read and Write Requests
+    conf.setIsChecksumEnabled(true);
+
+    Path testpath = new Path("a/b.txt");
+    String dataUploaded = "This is Sample Data";
+    FSDataOutputStream out = fs.create(testpath);
+    out.write(dataUploaded.getBytes(StandardCharsets.UTF_8));
+    out.hflush();
+    out.close();
+
+    FSDataInputStream in = fs.open(testpath);
+    byte[] bytesRead = new byte[dataUploaded.length()];
+    in.read(bytesRead);
+
+    // Verify that the data read is same as data written
+    Assertions.assertThat(bytesRead).describedAs("").containsExactly(dataUploaded.getBytes(StandardCharsets.UTF_8));
+    Assertions.assertThat(new String(bytesRead, StandardCharsets.UTF_8)).describedAs("").isEqualTo(dataUploaded);
+  }
+}