HADOOP-17990. Fix failing concurrent FS.initialize commands when fs.azure.createRemoteFileSystemDuringInitialization is enabled.

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

@@ -150,13 +150,7 @@ public void initialize(URI uri, Configuration configuration)
     if (abfsConfiguration.getCreateRemoteFileSystemDuringInitialization()) {
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.CREATE_FILESYSTEM, tracingHeaderFormat, listener);
-      if (this.tryGetFileStatus(new Path(AbfsHttpConstants.ROOT_PATH), tracingContext) == null) {
-        try {
-          this.createFileSystem(tracingContext);
-        } catch (AzureBlobFileSystemException ex) {
-          checkException(null, ex, AzureServiceErrorCode.FILE_SYSTEM_ALREADY_EXISTS);
-        }
-      }
+      this.createFileSystemIfNotExist(tracingContext);
     }
 
     LOG.trace("Initiate check for delegation token manager");
@@ -1411,6 +1405,30 @@ AzureBlobFileSystemStore getAbfsStore() {
     return abfsStore;
   }
 
+  @VisibleForTesting
+  void setAbfsStore(AzureBlobFileSystemStore abfsStore) {
+    this.abfsStore = abfsStore;
+  }
+
+  @VisibleForTesting
+  void createFileSystemIfNotExist(TracingContext tracingContext) throws IOException {
+    if (this.tryGetFileStatus(new Path(AbfsHttpConstants.ROOT_PATH), tracingContext) == null) {
+      try {
+        this.createFileSystem(tracingContext);
+      } catch (IOException ex) {
+        if (ex instanceof AzureBlobFileSystemException) {
+          checkException(null, (AzureBlobFileSystemException) ex,
+              AzureServiceErrorCode.FILE_SYSTEM_ALREADY_EXISTS);
+        } else if (ex.getCause() != null && ex.getCause() instanceof AzureBlobFileSystemException) {
+          checkException(null, (AzureBlobFileSystemException) ex.getCause(),
+              AzureServiceErrorCode.FILE_SYSTEM_ALREADY_EXISTS);
+        } else {
+          throw ex;
+        }
+      }
+    }
+  }
+
   @VisibleForTesting
   AbfsClient getAbfsClient() {
     return abfsStore.getClient();

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

@@ -19,9 +19,16 @@
 package org.apache.hadoop.fs.azurebfs;
 
 import java.io.FileNotFoundException;
+import java.io.IOException;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
+import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
+import org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode;
+import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
 import org.junit.Test;
 
 import org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys;
@@ -49,4 +56,39 @@ public void ensureFilesystemWillNotBeCreatedIfCreationConfigIsNotSet() throws Ex
     final AzureBlobFileSystem fs = this.createFileSystem();
     FileStatus[] fileStatuses = fs.listStatus(new Path("/"));
   }
+
+  @Test
+  public void ensureFilesystemWillBeCreatedIfCreationConfigIsSet() throws Exception {
+    final AzureBlobFileSystem fs = createFileSystem();
+    Configuration config = getRawConfiguration();
+    fs.initialize(FileSystem.getDefaultUri(config), config);
+
+    // Make sure createFileSystemIfNotExists is working as intended.
+    final MockAzureBlobFileSystemStore store = new MockAzureBlobFileSystemStore(config);
+    fs.setAbfsStore(store);
+    fs.createFileSystemIfNotExist(getTestTracingContext(fs, true));
+    assert(store.isCreateFileSystemCalled);
+  }
+
+  /**
+   * Mock AzureBlobFileSystemStore to simulate container already exists
+   * exception when calling createFileSystem command.
+   */
+  static class MockAzureBlobFileSystemStore extends AzureBlobFileSystemStore {
+    boolean isCreateFileSystemCalled = false;
+
+    public MockAzureBlobFileSystemStore(Configuration config) throws IOException {
+      super(FileSystem.getDefaultUri(config), true, config, null);
+    }
+
+    @Override
+    public void createFilesystem(TracingContext tracingContext) throws AzureBlobFileSystemException {
+      isCreateFileSystemCalled = true;
+      // Make sure createFileSystemIfNotExists works when the filesystem/container already exists.
+      throw new AbfsRestOperationException(
+          AzureServiceErrorCode.FILE_SYSTEM_ALREADY_EXISTS.getStatusCode(),
+          AzureServiceErrorCode.FILE_SYSTEM_ALREADY_EXISTS.getErrorCode(),
+          "This container is already exists", null);
+    }
+  }
 }