HADOOP-19604. ABFS: Full blob md5 computation during flush change to be config driven

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

@@ -438,6 +438,10 @@ public class AbfsConfiguration{
       FS_AZURE_ABFS_ENABLE_CHECKSUM_VALIDATION, DefaultValue = DEFAULT_ENABLE_ABFS_CHECKSUM_VALIDATION)
   private boolean isChecksumValidationEnabled;
 
+  @BooleanConfigurationValidatorAnnotation(ConfigurationKey =
+      FS_AZURE_ABFS_ENABLE_FULL_BLOB_CHECKSUM_VALIDATION, DefaultValue = DEFAULT_ENABLE_FULL_BLOB_ABFS_CHECKSUM_VALIDATION)
+  private boolean isFullBlobChecksumValidationEnabled;
+
   @BooleanConfigurationValidatorAnnotation(ConfigurationKey =
       FS_AZURE_ENABLE_PAGINATED_DELETE, DefaultValue = DEFAULT_ENABLE_PAGINATED_DELETE)
   private boolean isPaginatedDeleteEnabled;
@@ -1705,6 +1709,10 @@ public void setIsChecksumValidationEnabled(boolean isChecksumValidationEnabled)
     this.isChecksumValidationEnabled = isChecksumValidationEnabled;
   }
 
+  public boolean isFullBlobChecksumValidationEnabled() {
+    return isFullBlobChecksumValidationEnabled;
+  }
+
   public long getBlobCopyProgressPollWaitMillis() {
     return blobCopyProgressPollWaitMillis;
   }

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

@@ -356,6 +356,9 @@ public final class ConfigurationKeys {
   /** Add extra layer of verification of the integrity of the request content during transport: {@value}. */
   public static final String FS_AZURE_ABFS_ENABLE_CHECKSUM_VALIDATION = "fs.azure.enable.checksum.validation";
 
+  /** Add extra layer of verification of the integrity of the full blob request content during transport: {@value}. */
+  public static final String FS_AZURE_ABFS_ENABLE_FULL_BLOB_CHECKSUM_VALIDATION = "fs.azure.enable.full.blob.checksum.validation";
+
   public static String accountProperty(String property, String account) {
     return property + DOT + account;
   }

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

@@ -147,6 +147,7 @@ public final class FileSystemConfigurations {
   public static final boolean DEFAULT_ENABLE_ABFS_RENAME_RESILIENCE = true;
   public static final boolean DEFAULT_ENABLE_PAGINATED_DELETE = false;
   public static final boolean DEFAULT_ENABLE_ABFS_CHECKSUM_VALIDATION = false;
+  public static final boolean DEFAULT_ENABLE_FULL_BLOB_ABFS_CHECKSUM_VALIDATION = false;
 
   /**
    * Limit of queued block upload operations before writes

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

@@ -1073,11 +1073,15 @@ public AbfsRestOperation flush(byte[] buffer,
     requestHeaders.add(new AbfsHttpHeader(CONTENT_LENGTH, String.valueOf(buffer.length)));
     requestHeaders.add(new AbfsHttpHeader(CONTENT_TYPE, APPLICATION_XML));
     requestHeaders.add(new AbfsHttpHeader(IF_MATCH, eTag));
+    String md5Hash = null;
     if (leaseId != null) {
       requestHeaders.add(new AbfsHttpHeader(X_MS_LEASE_ID, leaseId));
     }
-    if (blobMd5 != null) {
+    if (isFullBlobChecksumValidationEnabled() && blobMd5 != null) {
       requestHeaders.add(new AbfsHttpHeader(X_MS_BLOB_CONTENT_MD5, blobMd5));
+    } else {
+      md5Hash = computeMD5Hash(buffer, 0, buffer.length);
+      requestHeaders.add(new AbfsHttpHeader(X_MS_BLOB_CONTENT_MD5, md5Hash));
     }
     final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_COMP, BLOCKLIST);
@@ -1103,8 +1107,21 @@ public AbfsRestOperation flush(byte[] buffer,
         AbfsRestOperation op1 = getPathStatus(path, true, tracingContext,
             contextEncryptionAdapter);
         String metadataMd5 = op1.getResult().getResponseHeader(CONTENT_MD5);
-        if (blobMd5 != null && !blobMd5.equals(metadataMd5)) {
-          throw ex;
+        /*
+         * Validate the response by comparing the server's MD5 metadata against either:
+         * 1. The full blob content MD5 (if full blob checksum validation is enabled), or
+         * 2. The full block ID buffer MD5 (fallback if blob checksum validation is disabled)
+         */
+        if (getAbfsConfiguration().isFullBlobChecksumValidationEnabled() && blobMd5 != null) {
+          // Full blob content MD5 mismatch — integrity check failed
+          if (!blobMd5.equals(metadataMd5)) {
+            throw ex;
+          }
+        } else {
+          // Block ID buffer MD5 mismatch — integrity check failed
+          if (md5Hash != null && !md5Hash.equals(metadataMd5)) {
+            throw ex;
+          }
         }
         return op;
       }

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

@@ -1423,6 +1423,17 @@ protected boolean isChecksumValidationEnabled() {
     return getAbfsConfiguration().getIsChecksumValidationEnabled();
   }
 
+  /**
+   * Conditions check for allowing checksum support for write operation.
+   * Server will support this if client sends the MD5 Hash as a request header.
+   * For azure stoage service documentation and more details refer to
+   * <a href="https://learn.microsoft.com/en-us/rest/api/storageservices/datalakestoragegen2/path/update">Path - Update Azure Rest API</a>.
+   * @return true if full blob checksum validation enabled.
+   */
+  protected boolean isFullBlobChecksumValidationEnabled() {
+    return getAbfsConfiguration().isFullBlobChecksumValidationEnabled();
+  }
+
   /**
    * Compute MD5Hash of the given byte array starting from given offset up to given length.
    * @param data byte array from which data is fetched to compute MD5 Hash.

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

@@ -867,10 +867,9 @@ public AbfsRestOperation flush(final String path,
     if (leaseId != null) {
       requestHeaders.add(new AbfsHttpHeader(X_MS_LEASE_ID, leaseId));
     }
-    if (isChecksumValidationEnabled() && blobMd5 != null) {
+    if (isFullBlobChecksumValidationEnabled() && blobMd5 != null) {
       requestHeaders.add(new AbfsHttpHeader(X_MS_BLOB_CONTENT_MD5, blobMd5));
     }
-
     final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_ACTION, FLUSH_ACTION);
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_POSITION, Long.toString(position));

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

@@ -223,7 +223,7 @@ public AbfsOutputStream(AbfsOutputStreamContext abfsOutputStreamContext)
       md5 = MessageDigest.getInstance(MD5);
       fullBlobContentMd5 = MessageDigest.getInstance(MD5);
     } catch (NoSuchAlgorithmException e) {
-      if (client.isChecksumValidationEnabled()) {
+      if (isChecksumValidationEnabled()) {
         throw new IOException("MD5 algorithm not available", e);
       }
     }
@@ -464,10 +464,13 @@ public synchronized void write(final byte[] data, final int off, final int lengt
     AbfsBlock block = createBlockIfNeeded(position);
     int written = bufferData(block, data, off, length);
     // Update the incremental MD5 hash with the written data.
-    getMessageDigest().update(data, off, written);
-
+    if (isChecksumValidationEnabled()) {
+      getMessageDigest().update(data, off, written);
+    }
     // Update the full blob MD5 hash with the written data.
-    getFullBlobContentMd5().update(data, off, written);
+    if (isFullBlobChecksumValidationEnabled()) {
+      getFullBlobContentMd5().update(data, off, written);
+    }
     int remainingCapacity = block.remainingCapacity();
 
     if (written < length) {
@@ -544,7 +547,12 @@ private void uploadBlockAsync(AbfsBlock blockToUpload,
     outputStreamStatistics.bytesToUpload(bytesLength);
     outputStreamStatistics.writeCurrentBuffer();
     DataBlocks.BlockUploadData blockUploadData = blockToUpload.startUpload();
-    String md5Hash = getMd5();
+    String md5Hash;
+    if (getClient().getAbfsConfiguration().getIsChecksumValidationEnabled()) {
+      md5Hash = getMd5();
+    } else {
+      md5Hash = null;
+    }
     final Future<Void> job =
         executorService.submit(() -> {
           AbfsPerfTracker tracker =
@@ -1222,6 +1230,20 @@ public MessageDigest getFullBlobContentMd5() {
     return fullBlobContentMd5;
   }
 
+  /**
+   * @return true if checksum validation is enabled.
+   */
+  public boolean isChecksumValidationEnabled() {
+    return getClient().isChecksumValidationEnabled();
+  }
+
+  /**
+   * @return true if full blob checksum validation is enabled.
+   */
+  public boolean isFullBlobChecksumValidationEnabled() {
+    return getClient().isFullBlobChecksumValidationEnabled();
+  }
+
   /**
    * Returns the Base64-encoded MD5 checksum based on the current digest state.
    * This finalizes the digest calculation. Returns null if the digest is empty.

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

@@ -95,7 +95,9 @@ protected synchronized AbfsBlock createBlockInternal(long position)
       setBlockCount(getBlockCount() + 1);
       AbfsBlock activeBlock = new AbfsBlobBlock(getAbfsOutputStream(), position, getBlockIdLength(), getBlockCount());
       activeBlock.setBlockEntry(addNewEntry(activeBlock.getBlockId(), activeBlock.getOffset()));
-      getAbfsOutputStream().getMessageDigest().reset();
+      if (getAbfsOutputStream().isChecksumValidationEnabled()) {
+        getAbfsOutputStream().getMessageDigest().reset();
+      }
       setActiveBlock(activeBlock);
     }
     return getActiveBlock();

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

@@ -180,7 +180,10 @@ protected synchronized AbfsRestOperation remoteFlush(final long offset,
       tracingContextFlush.setIngressHandler(BLOB_FLUSH);
       tracingContextFlush.setPosition(String.valueOf(offset));
       LOG.trace("Flushing data at offset {} for path {}", offset, getAbfsOutputStream().getPath());
-      String fullBlobMd5 = computeFullBlobMd5();
+      String fullBlobMd5 = null;
+      if (getClient().isFullBlobChecksumValidationEnabled()) {
+        fullBlobMd5 = computeFullBlobMd5();
+      }
       op = getClient().flush(blockListXml.getBytes(StandardCharsets.UTF_8),
           getAbfsOutputStream().getPath(),
           isClose, getAbfsOutputStream().getCachedSasTokenString(), leaseId,
@@ -194,7 +197,9 @@ isClose, getAbfsOutputStream().getCachedSasTokenString(), leaseId,
       LOG.error("Error in remote flush for path {} and offset {}", getAbfsOutputStream().getPath(), offset, ex);
       throw ex;
     } finally {
-      getAbfsOutputStream().getFullBlobContentMd5().reset();
+      if (getClient().isFullBlobChecksumValidationEnabled()) {
+        getAbfsOutputStream().getFullBlobContentMd5().reset();
+      }
     }
     return op;
   }
@@ -221,7 +226,7 @@ protected AbfsRestOperation remoteAppendBlobWrite(String path,
       AppendRequestParameters reqParams,
       TracingContext tracingContext) throws IOException {
     // Perform the remote append operation using the blob client.
-    AbfsRestOperation op = null;
+    AbfsRestOperation op;
     try {
       op = blobClient.appendBlock(path, reqParams, uploadData.toByteArray(), tracingContext);
     } catch (AbfsRestOperationException ex) {

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

@@ -62,7 +62,9 @@ protected synchronized AbfsBlock createBlockInternal(long position)
     if (getActiveBlock() == null) {
       setBlockCount(getBlockCount() + 1);
       AbfsBlock activeBlock = new AbfsBlock(getAbfsOutputStream(), position);
-      getAbfsOutputStream().getMessageDigest().reset();
+      if (getAbfsOutputStream().isChecksumValidationEnabled()) {
+        getAbfsOutputStream().getMessageDigest().reset();
+      }
       setActiveBlock(activeBlock);
     }
     return getActiveBlock();

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

@@ -179,7 +179,10 @@ protected synchronized AbfsRestOperation remoteFlush(final long offset,
       tracingContextFlush.setIngressHandler(DFS_FLUSH);
       tracingContextFlush.setPosition(String.valueOf(offset));
     }
-    String fullBlobMd5 = computeFullBlobMd5();
+    String fullBlobMd5 = null;
+    if (getClient().isFullBlobChecksumValidationEnabled()) {
+      fullBlobMd5 = computeFullBlobMd5();
+    }
     LOG.trace("Flushing data at offset {} and path {}", offset, getAbfsOutputStream().getPath());
     AbfsRestOperation op;
     try {
@@ -194,7 +197,9 @@ protected synchronized AbfsRestOperation remoteFlush(final long offset,
           getAbfsOutputStream().getPath(), offset, ex);
       throw ex;
     } finally {
-      getAbfsOutputStream().getFullBlobContentMd5().reset();
+      if (getClient().isFullBlobChecksumValidationEnabled()) {
+        getAbfsOutputStream().getFullBlobContentMd5().reset();
+      }
     }
     return op;
   }

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

@@ -211,13 +211,18 @@ void createRenamePendingJson(Path path, byte[] bytes)
     String blockId = generateBlockId();
     String blockList = generateBlockListXml(blockId);
     byte[] buffer = blockList.getBytes(StandardCharsets.UTF_8);
-    String computedMd5 = abfsClient.computeMD5Hash(buffer, 0,  buffer.length);
+    String computedMd5 = null;
+    if (abfsClient.isFullBlobChecksumValidationEnabled()) {
+      computedMd5 = abfsClient.computeMD5Hash(buffer, 0, buffer.length);
+    }
 
     AppendRequestParameters appendRequestParameters
         = new AppendRequestParameters(0, 0,
         bytes.length, AppendRequestParameters.Mode.APPEND_MODE, false, null,
         abfsClient.getAbfsConfiguration().isExpectHeaderEnabled(),
-        new BlobAppendRequestParameters(blockId, eTag), abfsClient.computeMD5Hash(bytes, 0,  bytes.length));
+        new BlobAppendRequestParameters(blockId, eTag),
+        abfsClient.isChecksumValidationEnabled() ? abfsClient.computeMD5Hash(
+            bytes, 0, bytes.length) : null);
 
     abfsClient.append(path.toUri().getPath(), bytes,
         appendRequestParameters, null, null, tracingContext);

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

@@ -2128,7 +2128,7 @@ private void testRenamePreRenameFailureResolution(final AzureBlobFileSystem fs)
               Mockito.anyBoolean(), Mockito.nullable(String.class),
               Mockito.nullable(String.class), Mockito.anyString(),
               Mockito.nullable(ContextEncryptionAdapter.class),
-              Mockito.any(TracingContext.class), Mockito.anyString());
+              Mockito.any(TracingContext.class), Mockito.nullable(String.class));
       return createAnswer.callRealMethod();
     };
     RenameAtomicityTestUtils.addCreatePathMock(client,

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

@@ -43,6 +43,7 @@
 
 import static java.net.HttpURLConnection.HTTP_CONFLICT;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ABFS_ENABLE_CHECKSUM_VALIDATION;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ABFS_ENABLE_FULL_BLOB_CHECKSUM_VALIDATION;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.assertDeleted;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.assertIsDirectory;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.assertMkdirs;
@@ -113,7 +114,10 @@ public void testReadFile() throws Exception {
     boolean[] createFileWithAbfs = new boolean[]{false, true, false, true};
     boolean[] readFileWithAbfs = new boolean[]{false, true, true, false};
 
-    AzureBlobFileSystem abfs = getFileSystem();
+    Configuration conf = getRawConfiguration();
+    conf.setBoolean(FS_AZURE_ABFS_ENABLE_FULL_BLOB_CHECKSUM_VALIDATION, true);
+    FileSystem fileSystem = FileSystem.newInstance(conf);
+    AzureBlobFileSystem abfs = (AzureBlobFileSystem) fileSystem;
     // test only valid for non-namespace enabled account
     Assume.assumeFalse("Namespace enabled account does not support this test",
         getIsNamespaceEnabled(abfs));
@@ -412,7 +416,10 @@ public void testScenario2() throws Exception {
    */
   @Test
   public void testScenario3() throws Exception {
-    AzureBlobFileSystem abfs = getFileSystem();
+    Configuration conf = getRawConfiguration();
+    conf.setBoolean(FS_AZURE_ABFS_ENABLE_FULL_BLOB_CHECKSUM_VALIDATION, true);
+    FileSystem fileSystem = FileSystem.newInstance(conf);
+    AzureBlobFileSystem abfs = (AzureBlobFileSystem) fileSystem;
     Assume.assumeFalse("Namespace enabled account does not support this test",
         getIsNamespaceEnabled(abfs));
     Assume.assumeFalse("Not valid for APPEND BLOB", isAppendBlobEnabled());

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

@@ -166,7 +166,12 @@ public static void setMockAbfsRestOperationForFlushOperation(
     requestHeaders.add(new AbfsHttpHeader(CONTENT_LENGTH, String.valueOf(buffer.length)));
     requestHeaders.add(new AbfsHttpHeader(CONTENT_TYPE, APPLICATION_XML));
     requestHeaders.add(new AbfsHttpHeader(IF_MATCH, eTag));
-    requestHeaders.add(new AbfsHttpHeader(X_MS_BLOB_CONTENT_MD5, blobMd5));
+    if (spiedClient.isFullBlobChecksumValidationEnabled()) {
+      requestHeaders.add(new AbfsHttpHeader(X_MS_BLOB_CONTENT_MD5, blobMd5));
+    } else {
+      requestHeaders.add(new AbfsHttpHeader(X_MS_BLOB_CONTENT_MD5,
+          spiedClient.computeMD5Hash(buffer, 0, buffer.length)));
+    }
     final AbfsUriQueryBuilder abfsUriQueryBuilder = spiedClient.createDefaultUriQueryBuilder();
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_COMP, BLOCKLIST);
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_CLOSE, String.valueOf(false));

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

@@ -427,7 +427,7 @@ public void testNoNetworkCallsForSecondFlush() throws Exception {
             Mockito.any(TracingContext.class));
     Mockito.verify(blobClient, Mockito.times(1)).
         flush(Mockito.any(byte[].class), Mockito.anyString(), Mockito.anyBoolean(), Mockito.any(), Mockito.any(), Mockito.anyString(), Mockito.any(),
-            Mockito.any(TracingContext.class), Mockito.anyString());
+            Mockito.any(TracingContext.class), Mockito.nullable(String.class));
   }
 
   /**
@@ -481,6 +481,8 @@ public void testResetCalledOnExceptionInRemoteFlush() throws Exception {
       //expected exception
     }
     // Verify that reset was called on the message digest
-    Mockito.verify(mockMessageDigest, Mockito.times(1)).reset();
+    if (spiedClient.isChecksumValidationEnabled()) {
+      Mockito.verify(mockMessageDigest, Mockito.times(1)).reset();
+    }
   }
 }