Refactor getAcl

Author: Anuj Modi

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

@@ -395,15 +395,21 @@ private synchronized boolean getNamespaceEnabledInformationFromServer(
     try {
       LOG.debug("Get root ACL status");
       getClient().getAclStatus(AbfsHttpConstants.ROOT_PATH, tracingContext);
+      // If getAcl succeeds, namespace is enabled.
       isNamespaceEnabled = Trilean.getTrilean(true);
     } catch (AbfsRestOperationException ex) {
-      // Get ACL status is a HEAD request, its response doesn't contain
-      // errorCode
+      // Get ACL status is a HEAD request, its response doesn't contain errorCode
       // So can only rely on its status code to determine its account type.
-      if (HttpURLConnection.HTTP_BAD_REQUEST != ex.getStatusCode()) {
+      if (HttpURLConnection.HTTP_BAD_REQUEST == ex.getStatusCode()) {
+        // If getAcl fails with 400, namespace is disabled.
+        isNamespaceEnabled = Trilean.getTrilean(false);
+      } else if (HttpURLConnection.HTTP_NOT_FOUND == ex.getStatusCode()) {
+        // If getAcl fails with 404, namespace is enabled.
+        isNamespaceEnabled = Trilean.getTrilean(true);
+      } else {
+        // Any other server error, throw exception.
         throw ex;
       }
-      isNamespaceEnabled = Trilean.getTrilean(false);
     } catch (AzureBlobFileSystemException ex) {
       throw ex;
     }

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

@@ -24,19 +24,23 @@
 import org.junit.Assume;
 import org.junit.Test;
 import org.assertj.core.api.Assertions;
+import org.mockito.Mockito;
 
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
 import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
 import org.apache.hadoop.fs.azurebfs.services.AbfsRestOperation;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.azurebfs.enums.Trilean;
 import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
+import org.apache.kerby.config.Conf;
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -48,6 +52,7 @@
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_IS_HNS_ENABLED;
 import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_TEST_NAMESPACE_ENABLED_ACCOUNT;
 import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+import static org.mockito.Mockito.when;
 
 /**
  * Test getIsNamespaceEnabled call.
@@ -217,4 +222,31 @@ private AbfsClient callAbfsGetIsNamespaceEnabledAndReturnMockAbfsClient()
     return mockClient;
   }
 
+  @Test
+  public void ensureGetAclDetermineHnsStatusAccurately() throws Exception {
+    ensureGetAclDetermineHnsStatusAccuratelyInternal(400, false);
+    ensureGetAclDetermineHnsStatusAccuratelyInternal(404, true);
+    ensureGetAclDetermineHnsStatusAccuratelyInternal(500, false);
+    ensureGetAclDetermineHnsStatusAccuratelyInternal(503, false);
+  }
+
+  private void ensureGetAclDetermineHnsStatusAccuratelyInternal(int statusCode,
+      boolean expected) throws Exception {
+    AzureBlobFileSystemStore store = Mockito.spy(getFileSystem().getAbfsStore());
+    AbfsClient mockClient = mock(AbfsClient.class);
+    store.setNamespaceEnabled(Trilean.UNKNOWN);
+    doReturn(mockClient).when(store).getClient();
+    AbfsRestOperationException ex = new AbfsRestOperationException(
+        statusCode, null, Integer.toString(statusCode), null);
+    doThrow(ex).when(mockClient).getAclStatus(anyString(), any(TracingContext.class));
+
+    try {
+      boolean isHnsEnabled = store.getIsNamespaceEnabled(
+          getTestTracingContext(getFileSystem(), false));
+      Assertions.assertThat(isHnsEnabled).isEqualTo(expected);
+    } catch (AbfsRestOperationException caughtEx) {
+      Assertions.assertThat(caughtEx.getStatusCode()).isEqualTo(statusCode);
+      Assertions.assertThat(caughtEx.getErrorMessage()).isEqualTo(ex.getErrorMessage());
+    }
+  }
 }