fix merge

Author: anmolanmol1234

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

@@ -403,6 +403,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;
@@ -1629,6 +1633,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

@@ -320,6 +320,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

@@ -139,6 +139,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

@@ -1100,9 +1100,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));
@@ -1127,7 +1128,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

@@ -1419,14 +1419,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();