HADOOP-17548. ABFS: Toggle Store Mkdirs request overwrite parameter (#2729) (#2781)

Contributed by Sumangala Patki.

(cherry picked from commit fe633d4739)

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

@@ -200,6 +200,11 @@ public class AbfsConfiguration{
       DefaultValue = DEFAULT_FS_AZURE_ENABLE_CONDITIONAL_CREATE_OVERWRITE)
   private boolean enableConditionalCreateOverwrite;
 
+  @BooleanConfigurationValidatorAnnotation(ConfigurationKey =
+      FS_AZURE_ENABLE_MKDIR_OVERWRITE, DefaultValue =
+      DEFAULT_FS_AZURE_ENABLE_MKDIR_OVERWRITE)
+  private boolean mkdirOverwrite;
+
   @StringConfigurationValidatorAnnotation(ConfigurationKey = FS_AZURE_APPEND_BLOB_KEY,
       DefaultValue = DEFAULT_FS_AZURE_APPEND_BLOB_DIRECTORIES)
   private String azureAppendBlobDirs;
@@ -633,6 +638,10 @@ public class AbfsConfiguration{
     return this.enableConditionalCreateOverwrite;
   }
 
+  public boolean isEnabledMkdirOverwrite() {
+    return mkdirOverwrite;
+  }
+
   public String getAppendBlobDirs() {
     return this.azureAppendBlobDirs;
   }

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

@@ -477,7 +477,7 @@ public class AzureBlobFileSystem extends FileSystem
       statIncrement(DIRECTORIES_CREATED);
       return true;
     } catch (AzureBlobFileSystemException ex) {
-      checkException(f, ex, AzureServiceErrorCode.PATH_ALREADY_EXISTS);
+      checkException(f, ex);
       return true;
     }
   }

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

@@ -634,7 +634,10 @@ public class AzureBlobFileSystemStore implements Closeable, ListingSupport {
               umask,
               isNamespaceEnabled);
 
-      final AbfsRestOperation op = client.createPath(getRelativePath(path), false, true,
+      boolean overwrite =
+          !isNamespaceEnabled || abfsConfiguration.isEnabledMkdirOverwrite();
+      final AbfsRestOperation op = client.createPath(getRelativePath(path),
+          false, overwrite,
               isNamespaceEnabled ? getOctalNotation(permission) : null,
               isNamespaceEnabled ? getOctalNotation(umask) : null, false, null);
       perfInfo.registerResult(op.getResult()).registerSuccess(true);

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

@@ -83,6 +83,7 @@ public final class ConfigurationKeys {
    *  overwritten only if there is a match on the eTag of existing file.
    */
   public static final String FS_AZURE_ENABLE_CONDITIONAL_CREATE_OVERWRITE = "fs.azure.enable.conditional.create.overwrite";
+  public static final String FS_AZURE_ENABLE_MKDIR_OVERWRITE = "fs.azure.enable.mkdir.overwrite";
   /** Provides a config to provide comma separated path prefixes on which Appendblob based files are created
    *  Default is empty. **/
   public static final String FS_AZURE_APPEND_BLOB_KEY = "fs.azure.appendblob.directories";

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

@@ -78,6 +78,7 @@ public final class FileSystemConfigurations {
 
   public static final String DEFAULT_FS_AZURE_ATOMIC_RENAME_DIRECTORIES = "/hbase";
   public static final boolean DEFAULT_FS_AZURE_ENABLE_CONDITIONAL_CREATE_OVERWRITE = true;
+  public static final boolean DEFAULT_FS_AZURE_ENABLE_MKDIR_OVERWRITE = true;
   public static final String DEFAULT_FS_AZURE_APPEND_BLOB_DIRECTORIES = "";
   public static final String DEFAULT_FS_AZURE_INFINITE_LEASE_DIRECTORIES = "";
   public static final int DEFAULT_LEASE_THREADS = 0;

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

@@ -43,6 +43,7 @@ public final class HttpHeaderConfigurations {
   public static final String USER_AGENT = "User-Agent";
   public static final String X_HTTP_METHOD_OVERRIDE = "X-HTTP-Method-Override";
   public static final String X_MS_CLIENT_REQUEST_ID = "x-ms-client-request-id";
+  public static final String X_MS_EXISTING_RESOURCE_TYPE = "x-ms-existing-resource-type";
   public static final String X_MS_DATE = "x-ms-date";
   public static final String X_MS_REQUEST_ID = "x-ms-request-id";
   public static final String X_MS_VERSION = "x-ms-version";

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

@@ -377,7 +377,18 @@ public class AbfsClient implements Closeable {
             HTTP_METHOD_PUT,
             url,
             requestHeaders);
+    try {
       op.execute();
+    } catch (AzureBlobFileSystemException ex) {
+      if (!isFile && op.getResult().getStatusCode() == HttpURLConnection.HTTP_CONFLICT) {
+        String existingResource =
+            op.getResult().getResponseHeader(X_MS_EXISTING_RESOURCE_TYPE);
+        if (existingResource != null && existingResource.equals(DIRECTORY)) {
+          return op; //don't throw ex on mkdirs for existing directory
+        }
+      }
+      throw ex;
+    }
     return op;
   }
 

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

@@ -20,15 +20,19 @@ package org.apache.hadoop.fs.azurebfs;
 
 import java.util.UUID;
 
+import org.junit.Assume;
 import org.junit.Test;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 
-import static org.apache.hadoop.fs.contract.ContractTestUtils.assertMkdirs;
-
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.CONNECTIONS_MADE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ENABLE_MKDIR_OVERWRITE;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DEFAULT_FS_AZURE_ENABLE_MKDIR_OVERWRITE;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.assertMkdirs;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 
 /**
  * Test mkdir operation.
@@ -41,12 +45,44 @@ public class ITestAzureBlobFileSystemMkDir extends AbstractAbfsIntegrationTest {
 
   @Test
   public void testCreateDirWithExistingDir() throws Exception {
+    Assume.assumeTrue(DEFAULT_FS_AZURE_ENABLE_MKDIR_OVERWRITE || !getFileSystem()
+        .getIsNamespaceEnabled());
     final AzureBlobFileSystem fs = getFileSystem();
     Path path = new Path("testFolder");
     assertMkdirs(fs, path);
     assertMkdirs(fs, path);
   }
 
+  @Test
+  public void testMkdirExistingDirOverwriteFalse() throws Exception {
+    Assume.assumeFalse("Ignore test until default overwrite is set to false",
+        DEFAULT_FS_AZURE_ENABLE_MKDIR_OVERWRITE);
+    Assume.assumeTrue("Ignore test for Non-HNS accounts",
+        getFileSystem().getIsNamespaceEnabled());
+    //execute test only for HNS account with default overwrite=false
+    Configuration config = new Configuration(this.getRawConfiguration());
+    config.set(FS_AZURE_ENABLE_MKDIR_OVERWRITE, Boolean.toString(false));
+    AzureBlobFileSystem fs = getFileSystem(config);
+    Path path = new Path("testFolder");
+    assertMkdirs(fs, path); //checks that mkdirs returns true
+    long timeCreated = fs.getFileStatus(path).getModificationTime();
+    assertMkdirs(fs, path); //call to existing dir should return success
+    assertEquals("LMT should not be updated for existing dir", timeCreated,
+        fs.getFileStatus(path).getModificationTime());
+  }
+
+  @Test
+  public void createDirWithExistingFilename() throws Exception {
+    Assume.assumeFalse("Ignore test until default overwrite is set to false",
+        DEFAULT_FS_AZURE_ENABLE_MKDIR_OVERWRITE && getFileSystem()
+            .getIsNamespaceEnabled());
+    final AzureBlobFileSystem fs = getFileSystem();
+    Path path = new Path("testFilePath");
+    fs.create(path);
+    assertTrue(fs.getFileStatus(path).isFile());
+    intercept(FileAlreadyExistsException.class, () -> fs.mkdirs(path));
+  }
+
   @Test
   public void testCreateRoot() throws Exception {
     assertMkdirs(getFileSystem(), new Path("/"));