HADOOP-17643 WASB : Make metadata checks case insensitive (#3103)

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

@@ -40,6 +40,7 @@ import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.Set;
 
@@ -181,6 +182,11 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
    */
   public static final String KEY_USE_LOCAL_SAS_KEY_MODE = "fs.azure.local.sas.key.mode";
 
+  /**
+   * Config to control case sensitive metadata key checks/retrieval. If this
+   * is false, blob metadata keys will be treated case insensitive.
+   */
+  private static final String KEY_BLOB_METADATA_KEY_CASE_SENSITIVE = "fs.azure.blob.metadata.key.case.sensitive";
   private static final String PERMISSION_METADATA_KEY = "hdi_permission";
   private static final String OLD_PERMISSION_METADATA_KEY = "asv_permission";
   private static final String IS_FOLDER_METADATA_KEY = "hdi_isfolder";
@@ -364,6 +370,8 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
 
   private String delegationToken;
 
+  private boolean metadataKeyCaseSensitive;
+
   /** The error message template when container is not accessible. */
   public static final String NO_ACCESS_TO_CONTAINER_MSG = "No credentials found for "
       + "account %s in the configuration, and its container %s is not "
@@ -585,6 +593,12 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
       LOG.warn("Unable to initialize HBase root as an atomic rename directory.");
     }
     LOG.debug("Atomic rename directories: {} ", setToString(atomicRenameDirs));
+    metadataKeyCaseSensitive = conf
+        .getBoolean(KEY_BLOB_METADATA_KEY_CASE_SENSITIVE, true);
+    if (!metadataKeyCaseSensitive) {
+      LOG.info("{} configured as false. Blob metadata will be treated case insensitive.",
+          KEY_BLOB_METADATA_KEY_CASE_SENSITIVE);
+    }
   }
 
   /**
@@ -1633,15 +1647,24 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
     blob.setMetadata(metadata);
   }
 
-  private static String getMetadataAttribute(CloudBlobWrapper blob,
+  private String getMetadataAttribute(HashMap<String, String> metadata,
       String... keyAlternatives) {
-    HashMap<String, String> metadata = blob.getMetadata();
     if (null == metadata) {
       return null;
     }
     for (String key : keyAlternatives) {
-      if (metadata.containsKey(key)) {
-        return metadata.get(key);
+      if (metadataKeyCaseSensitive) {
+        if (metadata.containsKey(key)) {
+          return metadata.get(key);
+        }
+      } else {
+        // See HADOOP-17643 for details on why this case insensitive metadata
+        // checks been added
+        for (Entry<String, String> entry : metadata.entrySet()) {
+          if (key.equalsIgnoreCase(entry.getKey())) {
+            return entry.getValue();
+          }
+        }
       }
     }
     return null;
@@ -1665,7 +1688,7 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
   }
 
   private PermissionStatus getPermissionStatus(CloudBlobWrapper blob) {
-    String permissionMetadataValue = getMetadataAttribute(blob,
+    String permissionMetadataValue = getMetadataAttribute(blob.getMetadata(),
         PERMISSION_METADATA_KEY, OLD_PERMISSION_METADATA_KEY);
     if (permissionMetadataValue != null) {
       return PermissionStatusJsonSerializer.fromJSONString(
@@ -1713,19 +1736,32 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
         OLD_LINK_BACK_TO_UPLOAD_IN_PROGRESS_METADATA_KEY);
   }
 
-  private static String getLinkAttributeValue(CloudBlobWrapper blob)
+  private String getLinkAttributeValue(CloudBlobWrapper blob)
       throws UnsupportedEncodingException {
-    String encodedLinkTarget = getMetadataAttribute(blob,
+    String encodedLinkTarget = getMetadataAttribute(blob.getMetadata(),
         LINK_BACK_TO_UPLOAD_IN_PROGRESS_METADATA_KEY,
         OLD_LINK_BACK_TO_UPLOAD_IN_PROGRESS_METADATA_KEY);
     return decodeMetadataAttribute(encodedLinkTarget);
   }
 
-  private static boolean retrieveFolderAttribute(CloudBlobWrapper blob) {
+  private boolean retrieveFolderAttribute(CloudBlobWrapper blob) {
     HashMap<String, String> metadata = blob.getMetadata();
-    return null != metadata
-        && (metadata.containsKey(IS_FOLDER_METADATA_KEY) || metadata
-            .containsKey(OLD_IS_FOLDER_METADATA_KEY));
+    if (null != metadata) {
+      if (metadataKeyCaseSensitive) {
+        return metadata.containsKey(IS_FOLDER_METADATA_KEY)
+            || metadata.containsKey(OLD_IS_FOLDER_METADATA_KEY);
+      } else {
+        // See HADOOP-17643 for details on why this case insensitive metadata
+        // checks been added
+        for (String key : metadata.keySet()) {
+          if (key.equalsIgnoreCase(IS_FOLDER_METADATA_KEY)
+              || key.equalsIgnoreCase(OLD_IS_FOLDER_METADATA_KEY)) {
+            return true;
+          }
+        }
+      }
+    }
+    return false;
   }
 
   private static void storeVersionAttribute(CloudBlobContainerWrapper container) {
@@ -1740,18 +1776,9 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
     container.setMetadata(metadata);
   }
 
-  private static String retrieveVersionAttribute(
-      CloudBlobContainerWrapper container) {
-    HashMap<String, String> metadata = container.getMetadata();
-    if (metadata == null) {
-      return null;
-    } else if (metadata.containsKey(VERSION_METADATA_KEY)) {
-      return metadata.get(VERSION_METADATA_KEY);
-    } else if (metadata.containsKey(OLD_VERSION_METADATA_KEY)) {
-      return metadata.get(OLD_VERSION_METADATA_KEY);
-    } else {
-      return null;
-    }
+  private String retrieveVersionAttribute(CloudBlobContainerWrapper container) {
+    return getMetadataAttribute(container.getMetadata(), VERSION_METADATA_KEY,
+        OLD_VERSION_METADATA_KEY);
   }
 
   @Override
@@ -2246,7 +2273,8 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
       CloudBlobWrapper blob = getBlobReference(key);
       blob.downloadAttributes(getInstrumentedContext());
 
-      String value = getMetadataAttribute(blob, ensureValidAttributeName(attribute));
+      String value = getMetadataAttribute(blob.getMetadata(),
+          ensureValidAttributeName(attribute));
       value = decodeMetadataAttribute(value);
       return value == null ? null : value.getBytes(METADATA_ENCODING);
     } catch (Exception e) {