Changes based on PR review

Author: bhattmanish98

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

@@ -1528,8 +1528,22 @@ void setMaxBackoffIntervalMilliseconds(int maxBackoffInterval) {
     this.maxBackoffInterval = maxBackoffInterval;
   }
 
+  /**
+   * Sets the namespace enabled account flag.
+   *
+   * @param isNamespaceEnabledAccount boolean value indicating if the account is namespace enabled.
+   */
+  void setIsNamespaceEnabledAccount(boolean isNamespaceEnabledAccount) {
+    this.isNamespaceEnabled = Trilean.getTrilean(isNamespaceEnabledAccount);
+  }
+
+  /**
+   * Sets the namespace enabled account flag for testing purposes.
+   *
+   * @param isNamespaceEnabledAccount Trilean value indicating if the account is namespace enabled.
+   */
   @VisibleForTesting
-  public void setIsNamespaceEnabledAccount(Trilean isNamespaceEnabledAccount) {
+  void setIsNamespaceEnabledAccountForTesting(Trilean isNamespaceEnabledAccount) {
     this.isNamespaceEnabled = isNamespaceEnabledAccount;
   }
 

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

@@ -394,28 +394,28 @@ public boolean getIsNamespaceEnabled(TracingContext tracingContext)
 
   private synchronized boolean getNamespaceEnabledInformationFromServer(
       final TracingContext tracingContext) throws AzureBlobFileSystemException {
-    if (abfsConfiguration.getIsNamespaceEnabledAccount() != Trilean.UNKNOWN) {
+    if (getAbfsConfiguration().getIsNamespaceEnabledAccount() != Trilean.UNKNOWN) {
       return isNamespaceEnabled();
     }
     try {
       LOG.debug("Get root ACL status");
       getClient(AbfsServiceType.DFS).getAclStatus(AbfsHttpConstants.ROOT_PATH, tracingContext);
       // If getAcl succeeds, namespace is enabled.
-      setNamespaceEnabled(Trilean.TRUE);
+      setNamespaceEnabled(true);
     } catch (AbfsRestOperationException ex) {
       // Get ACL status is a HEAD request, its response doesn't contain errorCode
       // So can only rely on its status code to determine account type.
       if (HttpURLConnection.HTTP_BAD_REQUEST != ex.getStatusCode()) {
         // If getAcl fails with anything other than 400, namespace is enabled.
-        setNamespaceEnabled(Trilean.TRUE);
+        setNamespaceEnabled(true);
         // Continue to throw exception as earlier.
         LOG.debug("Failed to get ACL status with non 400. Inferring namespace enabled", ex);
         throw ex;
       }
       // If getAcl fails with 400, namespace is disabled.
       LOG.debug("Failed to get ACL status with 400. "
           + "Inferring namespace disabled and ignoring error", ex);
-      setNamespaceEnabled(Trilean.FALSE);
+      setNamespaceEnabled(false);
     } catch (AzureBlobFileSystemException ex) {
       throw ex;
     }
@@ -428,7 +428,7 @@ private synchronized boolean getNamespaceEnabledInformationFromServer(
    */
   @VisibleForTesting
   boolean isNamespaceEnabled() throws TrileanConversionException {
-    return abfsConfiguration.getIsNamespaceEnabledAccount().toBoolean();
+    return getAbfsConfiguration().getIsNamespaceEnabledAccount().toBoolean();
   }
 
   @VisibleForTesting
@@ -2011,9 +2011,8 @@ DataBlocks.BlockFactory getBlockFactory() {
     return blockFactory;
   }
 
-  @VisibleForTesting
-  void setNamespaceEnabled(Trilean isNamespaceEnabled){
-    abfsConfiguration.setIsNamespaceEnabledAccount(isNamespaceEnabled);
+  void setNamespaceEnabled(boolean isNamespaceEnabled){
+    getAbfsConfiguration().setIsNamespaceEnabledAccount(isNamespaceEnabled);
   }
 
   @VisibleForTesting

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

@@ -1720,10 +1720,10 @@ protected String getUserAgent() {
    */
   public boolean getIsNamespaceEnabled() throws AbfsDriverException {
     try {
-      return abfsConfiguration.getIsNamespaceEnabledAccount().toBoolean();
+      return getAbfsConfiguration().getIsNamespaceEnabledAccount().toBoolean();
     } catch (TrileanConversionException ex) {
       LOG.error("Failed to convert namespace enabled account property to boolean", ex);
-      throw new AbfsDriverException("Failed to determine if namespace is enabled", ex);
+      throw new AbfsDriverException("Failed to determine account type", ex);
     }
   }
 

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

@@ -102,8 +102,8 @@ private void setTestUserFs() throws Exception {
     conf.setBoolean(FS_AZURE_ACCOUNT_IS_HNS_ENABLED, isHNSEnabled);
     this.testUserFs = FileSystem.newInstance(conf);
     // Resetting the namespace enabled flag to unknown after file system init.
-    ((AzureBlobFileSystem) testUserFs).getAbfsStore().setNamespaceEnabled(
-        Trilean.UNKNOWN);
+    ((AzureBlobFileSystem) testUserFs).getAbfsStore()
+        .getAbfsConfiguration().setIsNamespaceEnabledAccountForTesting(Trilean.UNKNOWN);
   }
 
   private void setTestFsConf(final String fsConfKey,

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

@@ -82,7 +82,7 @@ public void testGetAclCallOnHnsConfigAbsence() throws Exception {
     AzureBlobFileSystemStore store = Mockito.spy(fs.getAbfsStore());
     AbfsClient client = Mockito.spy(fs.getAbfsStore().getClient(AbfsServiceType.DFS));
     Mockito.doReturn(client).when(store).getClient(AbfsServiceType.DFS);
-    store.setNamespaceEnabled(Trilean.UNKNOWN);
+    store.getAbfsConfiguration().setIsNamespaceEnabledAccountForTesting(Trilean.UNKNOWN);
 
     TracingContext tracingContext = getSampleTracingContext(fs, true);
     Mockito.doReturn(Mockito.mock(AbfsRestOperation.class))

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

@@ -29,6 +29,7 @@
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.azurebfs.constants.AbfsServiceType;
+import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsDriverException;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
 import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
 import org.apache.hadoop.fs.azurebfs.services.AbfsDfsClient;
@@ -159,7 +160,8 @@ public void testFailedRequestWhenFSNotExist() throws Exception {
             + testUri.substring(testUri.indexOf("@"));
     config.setBoolean(FS_AZURE_ACCOUNT_IS_HNS_ENABLED, isUsingXNSAccount);
     AzureBlobFileSystem fs = this.getFileSystem(nonExistingFsUrl);
-    fs.getAbfsStore().setNamespaceEnabled(Trilean.UNKNOWN);
+    fs.getAbfsStore().getAbfsConfiguration()
+        .setIsNamespaceEnabledAccountForTesting(Trilean.UNKNOWN);
 
     FileNotFoundException ex = intercept(FileNotFoundException.class, ()-> {
       fs.getFileStatus(new Path("/")); // Run a dummy FS call
@@ -207,7 +209,8 @@ public void testEnsureGetAclCallIsMadeOnceWhenConfigIsNotPresent()
   private void ensureGetAclCallIsMadeOnceForInvalidConf(String invalidConf)
       throws Exception {
     this.getFileSystem().getAbfsStore()
-        .setNamespaceEnabled(Trilean.getTrilean(invalidConf));
+        .getAbfsConfiguration()
+        .setIsNamespaceEnabledAccountForTesting(Trilean.getTrilean(invalidConf));
     AbfsClient mockClient =
         callAbfsGetIsNamespaceEnabledAndReturnMockAbfsClient();
     verify(mockClient, times(1))
@@ -217,15 +220,17 @@ private void ensureGetAclCallIsMadeOnceForInvalidConf(String invalidConf)
   private void ensureGetAclCallIsNeverMadeForValidConf(String validConf)
       throws Exception {
     this.getFileSystem().getAbfsStore()
-        .setNamespaceEnabled(Trilean.getTrilean(validConf));
+        .getAbfsConfiguration()
+        .setIsNamespaceEnabledAccountForTesting(Trilean.getTrilean(validConf));
     AbfsClient mockClient =
         callAbfsGetIsNamespaceEnabledAndReturnMockAbfsClient();
     verify(mockClient, never())
         .getAclStatus(anyString(), any(TracingContext.class));
   }
 
   private void unsetConfAndEnsureGetAclCallIsMadeOnce() throws IOException {
-    this.getFileSystem().getAbfsStore().setNamespaceEnabled(Trilean.UNKNOWN);
+    this.getFileSystem().getAbfsStore()
+        .getAbfsConfiguration().setIsNamespaceEnabledAccountForTesting(Trilean.UNKNOWN);
     AbfsClient mockClient =
         callAbfsGetIsNamespaceEnabledAndReturnMockAbfsClient();
     verify(mockClient, times(1))
@@ -262,7 +267,7 @@ private void ensureGetAclDetermineHnsStatusAccuratelyInternal(int statusCode,
       boolean expectedValue, boolean isExceptionExpected) throws Exception {
     AzureBlobFileSystemStore store = Mockito.spy(getFileSystem().getAbfsStore());
     AbfsClient mockClient = mock(AbfsClient.class);
-    store.setNamespaceEnabled(Trilean.UNKNOWN);
+    store.getAbfsConfiguration().setIsNamespaceEnabledAccountForTesting(Trilean.UNKNOWN);
     doReturn(mockClient).when(store).getClient(AbfsServiceType.DFS);
     AbfsRestOperationException ex = new AbfsRestOperationException(
         statusCode, null, Integer.toString(statusCode), null);
@@ -354,9 +359,7 @@ public void testAccountSpecificConfig() throws Exception {
    */
   @Test
   public void testNameSpaceConfig() throws Exception {
-    Configuration configuration = getRawConfiguration();
-    configuration.unset(FS_AZURE_ACCOUNT_IS_HNS_ENABLED);
-    configuration.unset(accountProperty(FS_AZURE_ACCOUNT_IS_HNS_ENABLED, this.getAccountName()));
+    Configuration configuration = getConfigurationWithoutHnsConfig();
     AzureBlobFileSystem abfs = (AzureBlobFileSystem) FileSystem.newInstance(configuration);
     AbfsConfiguration abfsConfig = new AbfsConfiguration(configuration, "bogusAccountName");
 
@@ -414,6 +417,63 @@ public void testNameSpaceConfig() throws Exception {
         .isEqualTo(false);
   }
 
+  /**
+   * Tests to check that the namespace configuration is set correctly
+   * during the initialization of the AzureBlobFileSystem.
+   *
+   *
+   * @throws Exception if any error occurs during configuration setup or evaluation
+   */
+  @Test
+  public void testFsInitShouldSetNamespaceConfig() throws Exception {
+    // Mock the AzureBlobFileSystem and its dependencies
+    AzureBlobFileSystem mockFileSystem = Mockito.spy((AzureBlobFileSystem)
+        FileSystem.newInstance(getConfigurationWithoutHnsConfig()));
+    AzureBlobFileSystemStore mockStore = Mockito.spy(mockFileSystem.getAbfsStore());
+    AbfsClient abfsClient = Mockito.spy(mockStore.getClient());
+    Mockito.doReturn(abfsClient).when(mockStore).getClient();
+    Mockito.doReturn(abfsClient).when(mockStore).getClient(any());
+    abfsClient.getIsNamespaceEnabled();
+    // Verify that getAclStatus is called once during initialization
+    Mockito.verify(abfsClient, times(0))
+        .getAclStatus(anyString(), any(TracingContext.class));
+
+    mockStore.getAbfsConfiguration().setIsNamespaceEnabledAccountForTesting(Trilean.UNKNOWN);
+    // In case someone wrongly configured the namespace in between the processing,
+    // abfsclient.getIsNamespaceEnabled() should throw an exception.
+    String errorMessage = intercept(AbfsDriverException.class, () -> {
+      abfsClient.getIsNamespaceEnabled();
+    }).getMessage();
+    Assertions.assertThat(errorMessage)
+        .describedAs("Client should throw exception when namespace is unknown")
+        .contains("Failed to determine account type");
+
+    // In case of unknown namespace, store's getIsNamespaceEnabled should call getAclStatus
+    // to determine the namespace status.
+    mockStore.getIsNamespaceEnabled(getTestTracingContext(mockFileSystem, false));
+    Mockito.verify(abfsClient, times(1))
+        .getAclStatus(anyString(), any(TracingContext.class));
+
+    abfsClient.getIsNamespaceEnabled();
+    // Verify that client's getNamespaceEnabled will not call getAclStatus again
+    Mockito.verify(abfsClient, times(1))
+        .getAclStatus(anyString(), any(TracingContext.class));
+  }
+
+  /**
+   * Returns the configuration without the HNS config set.
+   *
+   * @return Configuration without HNS config
+   */
+  private Configuration getConfigurationWithoutHnsConfig() {
+    Configuration configuration = getRawConfiguration();
+    configuration.unset(FS_AZURE_ACCOUNT_IS_HNS_ENABLED);
+    configuration.unset(accountProperty(FS_AZURE_ACCOUNT_IS_HNS_ENABLED, this.getAccountName()));
+    configuration.setBoolean(AZURE_CREATE_REMOTE_FILESYSTEM_DURING_INITIALIZATION, true);
+    configuration.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, getTestUrl());
+    return configuration;
+  }
+
   private void assertFileSystemInitWithExpectedHNSSettings(
       Configuration configuration, boolean expectedIsHnsEnabledValue) throws IOException {
     try (AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(configuration)) {

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

@@ -200,15 +200,15 @@ public void testExternalOps() throws Exception {
         0));
 
     // unset namespaceEnabled to call getAcl -> trigger tracing header validator
-    fs.getAbfsStore().setNamespaceEnabled(Trilean.UNKNOWN);
+    fs.getAbfsStore().getAbfsConfiguration().setIsNamespaceEnabledAccountForTesting(Trilean.UNKNOWN);
     fs.hasPathCapability(new Path("/"), CommonPathCapabilities.FS_ACLS);
 
     Assume.assumeTrue(getIsNamespaceEnabled(getFileSystem()));
     Assume.assumeTrue(getConfiguration().isCheckAccessEnabled());
     Assume.assumeTrue(getAuthType() == AuthType.OAuth);
 
     fs.setListenerOperation(FSOperationType.ACCESS);
-    fs.getAbfsStore().setNamespaceEnabled(Trilean.TRUE);
+    fs.getAbfsStore().getAbfsConfiguration().setIsNamespaceEnabledAccountForTesting(Trilean.TRUE);
     fs.access(new Path("/"), FsAction.READ);
   }