fail fs init if cpk config on non hns account

Author: saxenapranav

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

@@ -44,6 +44,7 @@
 
 import javax.annotation.Nullable;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.fs.impl.BackReference;
 import org.apache.hadoop.security.ProviderUtils;
@@ -211,6 +212,17 @@ public void initialize(URI uri, Configuration configuration)
     tracingHeaderFormat = abfsConfiguration.getTracingHeaderFormat();
     this.setWorkingDirectory(this.getHomeDirectory());
 
+    if (!getIsNamespaceEnabled(
+        new TracingContext(clientCorrelationId, fileSystemId,
+            FSOperationType.GET_FILESTATUS, tracingHeaderFormat, listener)) && (
+        abfsConfiguration.createEncryptionContextProvider() != null
+            || StringUtils.isNotEmpty(
+            abfsConfiguration.getEncodedClientProvidedEncryptionKey()))) {
+      close();
+      throw new PathIOException(
+          "Non HNS account " + uri.getPath() + " can not have CPK configs enabled.");
+    }
+
     TracingContext tracingContext = new TracingContext(clientCorrelationId,
             fileSystemId, FSOperationType.CREATE_FILESYSTEM, tracingHeaderFormat, listener);
     if (abfsConfiguration.getCreateRemoteFileSystemDuringInitialization()) {

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

@@ -40,7 +40,6 @@
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsInvalidChecksumException;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsDriverException;
 import org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode;
-import org.apache.hadoop.fs.azurebfs.utils.NamespaceUtil;
 import org.apache.hadoop.fs.store.LogExactlyOnce;
 import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystemStore.Permissions;
 import org.apache.hadoop.fs.azurebfs.extensions.EncryptionContextProvider;
@@ -120,7 +119,6 @@ public class AbfsClient implements Closeable {
   private final AbfsThrottlingIntercept intercept;
 
   private final ListeningScheduledExecutorService executorService;
-  private Boolean isNamespaceEnabled;
 
   private boolean renameResilience;
 
@@ -305,9 +303,6 @@ private void addEncryptionKeyRequestHeaders(String path,
       List<AbfsHttpHeader> requestHeaders, boolean isCreateFileRequest,
       ContextEncryptionAdapter contextEncryptionAdapter, TracingContext tracingContext)
       throws AzureBlobFileSystemException {
-    if (!getIsNamespaceEnabled(tracingContext)) {
-      return;
-    }
     String encodedKey, encodedKeySHA256;
     switch (encryptionType) {
     case GLOBAL_KEY:
@@ -1486,15 +1481,6 @@ public synchronized String getAccessToken() throws IOException {
     }
   }
 
-  private synchronized Boolean getIsNamespaceEnabled(TracingContext tracingContext)
-      throws AzureBlobFileSystemException {
-    if (isNamespaceEnabled == null) {
-      setIsNamespaceEnabled(NamespaceUtil.isNamespaceEnabled(this,
-          tracingContext));
-    }
-    return isNamespaceEnabled;
-  }
-
   protected Boolean getIsPaginatedDeleteEnabled() {
     return abfsConfiguration.isPaginatedDeleteEnabled();
   }
@@ -1684,11 +1670,6 @@ void setEncryptionContextProvider(EncryptionContextProvider provider) {
     encryptionContextProvider = provider;
   }
 
-  @VisibleForTesting
-  void setIsNamespaceEnabled(final Boolean isNamespaceEnabled) {
-    this.isNamespaceEnabled = isNamespaceEnabled;
-  }
-
   /**
    * Getter for abfsCounters from AbfsClient.
    * @return AbfsCounters instance.

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

@@ -209,8 +209,6 @@ public void setup() throws Exception {
       wasb = new NativeAzureFileSystem(azureNativeFileSystemStore);
       wasb.initialize(wasbUri, rawConfig);
     }
-    // Todo: To be fixed in HADOOP-19137
-    AbfsClientUtils.setIsNamespaceEnabled(abfs.getAbfsClient(), true);
   }
 
   @After

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

@@ -281,7 +281,6 @@ public void testCreateFileOverwrite(boolean enableConditionalCreateOverwrite)
     final AzureBlobFileSystem fs =
         (AzureBlobFileSystem) FileSystem.newInstance(currentFs.getUri(),
             config);
-    AbfsClientUtils.setIsNamespaceEnabled(fs.getAbfsClient(), true);
 
     long totalConnectionMadeBeforeTest = fs.getInstrumentationMap()
         .get(CONNECTIONS_MADE.getStatName());

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

@@ -26,9 +26,6 @@ public final class AbfsClientUtils {
   private AbfsClientUtils() {
 
   }
-  public static void setIsNamespaceEnabled(final AbfsClient abfsClient, final Boolean isNamespaceEnabled) {
-    abfsClient.setIsNamespaceEnabled(isNamespaceEnabled);
-  }
 
   public static void setEncryptionContextProvider(final AbfsClient abfsClient, final EncryptionContextProvider provider) {
     abfsClient.setEncryptionContextProvider(provider);