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_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_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

@@ -1076,9 +1076,10 @@ public AbfsRestOperation flush(byte[] buffer,
     if (leaseId != null) {
       requestHeaders.add(new AbfsHttpHeader(X_MS_LEASE_ID, leaseId));
     }
-    if (blobMd5 != null) {
-      requestHeaders.add(new AbfsHttpHeader(X_MS_BLOB_CONTENT_MD5, blobMd5));
-    }
+    String md5Value = (isFullBlobChecksumValidationEnabled() && blobMd5 != null)
+        ? blobMd5
+        : computeMD5Hash(buffer, 0, buffer.length);
+    requestHeaders.add(new AbfsHttpHeader(X_MS_BLOB_CONTENT_MD5, md5Value));
     final AbfsUriQueryBuilder abfsUriQueryBuilder = createDefaultUriQueryBuilder();
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_COMP, BLOCKLIST);
     abfsUriQueryBuilder.addQuery(QUERY_PARAM_CLOSE, String.valueOf(isClose));
@@ -1103,7 +1104,12 @@ 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)) {
+        /*
+         * 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 list buffer MD5 (fallback if blob checksum validation is disabled)
+         */
+        if (md5Value != null && !md5Value.equals(metadataMd5)) {
           throw ex;
         }
         return op;

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

@@ -1415,14 +1415,25 @@ protected boolean isChecksumValidationEnabled(List<AbfsHttpHeader> requestHeader
   /**
    * 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
+   * For azure storage 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 checksum validation enabled.
    */
   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 storage 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,7 @@ private void uploadBlockAsync(AbfsBlock blockToUpload,
     outputStreamStatistics.bytesToUpload(bytesLength);
     outputStreamStatistics.writeCurrentBuffer();
     DataBlocks.BlockUploadData blockUploadData = blockToUpload.startUpload();
-    String md5Hash = getMd5();
+    String md5Hash = getClient().isChecksumValidationEnabled() ? getMd5() : null;
     final Future<Void> job =
         executorService.submit(() -> {
           AbfsPerfTracker tracker =
@@ -1222,6 +1225,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

@@ -2125,7 +2125,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,