HADOOP-19658_3.4. ABFS:Create and rename idempotency for FNS Blob (#7914) (#7929)

Contributed by Anmol Asrani

Author: anmolanmol1234

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

@@ -457,6 +457,10 @@ public class AbfsConfiguration{
  DefaultValue = DEFAULT_FS_AZURE_ENABLE_CLIENT_TRANSACTION_ID)
  private boolean enableClientTransactionId;
 
+ @BooleanConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_ENABLE_CREATE_BLOB_IDEMPOTENCY,
+ DefaultValue = DEFAULT_FS_AZURE_ENABLE_CREATE_BLOB_IDEMPOTENCY)
+ private boolean enableCreateIdempotency;
+
  private String clientProvidedEncryptionKey;
  private String clientProvidedEncryptionKeySHA;
 
@@ -1005,6 +1009,12 @@ public String getAzureAtomicRenameDirs() {
  }
 
  public boolean isConditionalCreateOverwriteEnabled() {
+ // If either the configured FS service type or the ingress service type is BLOB,
+ // conditional create-overwrite is not used.
+ if (getIsCreateIdempotencyEnabled() && (getFsConfiguredServiceType() == AbfsServiceType.BLOB
+ || getIngressServiceType() == AbfsServiceType.BLOB)) {
+ return false;
+ }
  return this.enableConditionalCreateOverwrite;
  }
 
@@ -1136,6 +1146,10 @@ public boolean getIsClientTransactionIdEnabled() {
  return enableClientTransactionId;
  }
 
+ public boolean getIsCreateIdempotencyEnabled() {
+ return enableCreateIdempotency;
+ }
+
  /**
  * Enum config to allow user to pick format of x-ms-client-request-id header
  * @return tracingContextFormat config if valid, else default ALL_ID_FORMAT

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

@@ -394,6 +394,8 @@ public static String containerProperty(String property, String fsName, String ac
  public static final String FS_AZURE_BLOB_DIR_DELETE_MAX_THREAD = "fs.azure.blob.dir.delete.max.thread";
  /**Flag to enable/disable sending client transactional ID during create/rename operations: {@value}*/
  public static final String FS_AZURE_ENABLE_CLIENT_TRANSACTION_ID = "fs.azure.enable.client.transaction.id";
+ /**Flag to enable/disable create idempotency during create operation: {@value}*/
+ public static final String FS_AZURE_ENABLE_CREATE_BLOB_IDEMPOTENCY = "fs.azure.enable.create.blob.idempotency";
 
  private ConfigurationKeys() {}
 }

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

@@ -232,5 +232,7 @@ public final class FileSystemConfigurations {
 
  public static final boolean DEFAULT_FS_AZURE_ENABLE_CLIENT_TRANSACTION_ID = true;
 
+ public static final boolean DEFAULT_FS_AZURE_ENABLE_CREATE_BLOB_IDEMPOTENCY = true;
+
  private FileSystemConfigurations() {}
 }

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

@@ -509,9 +509,34 @@ public AbfsRestOperation createPath(final String path,
  final TracingContext tracingContext) throws AzureBlobFileSystemException {
  AbfsRestOperation op;
  if (isFileCreation) {
- // Create a file with the specified parameters
- op = createFile(path, overwrite, permissions, isAppendBlob, eTag,
- contextEncryptionAdapter, tracingContext);
+ if (getAbfsConfiguration().getIsCreateIdempotencyEnabled()) {
+ AbfsRestOperation statusOp = null;
+ try {
+ // Check if the file already exists by calling GetPathStatus
+ statusOp = getPathStatus(path, tracingContext, null, false);
+ } catch (AbfsRestOperationException ex) {
+ // If the path does not exist, continue with file creation
+ // For other errors, rethrow the exception
+ if (ex.getStatusCode() != HTTP_NOT_FOUND) {
+ throw ex;
+ }
+ }
+ // If the file exists and overwrite is not allowed, throw conflict
+ if (statusOp != null && statusOp.hasResult() && !overwrite) {
+ throw new AbfsRestOperationException(
+ HTTP_CONFLICT,
+ AzureServiceErrorCode.PATH_CONFLICT.getErrorCode(),
+ PATH_EXISTS,
+ null);
+ } else {
+ // Proceed with file creation (force overwrite = true)
+ op = createFile(path, true, permissions, isAppendBlob, eTag,
+ contextEncryptionAdapter, tracingContext);
+ }
+ } else {
+ op = createFile(path, overwrite, permissions, isAppendBlob, eTag,
+ contextEncryptionAdapter, tracingContext);
+ }
  } else {
  // Create a directory with the specified parameters
  op = createDirectory(path, permissions, isAppendBlob, eTag,
@@ -584,7 +609,6 @@ public AbfsRestOperation createPathRestOp(final String path,
  if (eTag != null && !eTag.isEmpty()) {
  requestHeaders.add(new AbfsHttpHeader(HttpHeaderConfigurations.IF_MATCH, eTag));
  }
-
  final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
  final AbfsRestOperation op = getAbfsRestOperation(
  AbfsRestOperationType.PutBlob,

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

@@ -18,8 +18,10 @@
 
 package org.apache.hadoop.fs.azurebfs.services;
 
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -54,7 +56,15 @@ public static void dumpHeadersToDebugLog(final String origin,
  if (key == null) {
  key = "HTTP Response";
  }
- String values = StringUtils.join(";", entry.getValue());
+ List<String> valuesList = entry.getValue();
+ if (valuesList == null) {
+ valuesList = Collections.emptyList();
+ } else {
+ valuesList = valuesList.stream()
+ .map(v -> v == null ? "" : v) // replace null with empty string
+ .collect(Collectors.toList());
+ }
+ String values = StringUtils.join(";", valuesList);
  if (key.contains("Cookie")) {
  values = "*cookie info*";
  }