HADOOP-16136. ABFS: Should only transform username to short name

Contributed by Da Zhou.

(cherry picked from commit 3988e75ca385aec31ca1fc49d6cffce1ea935825)
Signed-off-by: Steve Loughran <stevel@apache.org>

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

@@ -491,10 +491,12 @@ public class AzureBlobFileSystemStore {
 
     final String transformedOwner = identityTransformer.transformIdentityForGetRequest(
               result.getResponseHeader(HttpHeaderConfigurations.X_MS_OWNER),
+              true,
               userName);
 
     final String transformedGroup = identityTransformer.transformIdentityForGetRequest(
               result.getResponseHeader(HttpHeaderConfigurations.X_MS_GROUP),
+              false,
               primaryUserGroup);
 
     return new VersionedFileStatus(
@@ -536,8 +538,8 @@ public class AzureBlobFileSystemStore {
       long blockSize = abfsConfiguration.getAzureBlockSize();
 
       for (ListResultEntrySchema entry : retrievedSchema.paths()) {
-        final String owner = identityTransformer.transformIdentityForGetRequest(entry.owner(), userName);
-        final String group = identityTransformer.transformIdentityForGetRequest(entry.group(), primaryUserGroup);
+        final String owner = identityTransformer.transformIdentityForGetRequest(entry.owner(), true, userName);
+        final String group = identityTransformer.transformIdentityForGetRequest(entry.group(), false, primaryUserGroup);
         final FsPermission fsPermission = entry.permissions() == null
                 ? new AbfsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL)
                 : AbfsPermission.valueOf(entry.permissions());
@@ -758,9 +760,11 @@ public class AzureBlobFileSystemStore {
 
     final String transformedOwner = identityTransformer.transformIdentityForGetRequest(
             result.getResponseHeader(HttpHeaderConfigurations.X_MS_OWNER),
+            true,
             userName);
     final String transformedGroup = identityTransformer.transformIdentityForGetRequest(
             result.getResponseHeader(HttpHeaderConfigurations.X_MS_GROUP),
+            false,
             primaryUserGroup);
 
     final String permissions = result.getResponseHeader(HttpHeaderConfigurations.X_MS_PERMISSIONS);

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

@@ -80,53 +80,54 @@ public class IdentityTransformer {
   /**
    * Perform identity transformation for the Get request results in AzureBlobFileSystemStore:
    * getFileStatus(), listStatus(), getAclStatus().
-   * Input originalUserOrGroup can be one of the following:
+   * Input originalIdentity can be one of the following:
    * 1. $superuser:
    *     by default it will be transformed to local user/group, this can be disabled by setting
    *     "fs.azure.identity.transformer.skip.superuser.replacement" to true.
    *
    * 2. User principal id:
-   *     can be transformed to localUserOrGroup, if this principal id matches the principal id set in
-   *     "fs.azure.identity.transformer.service.principal.id" and localUserOrGroup is stated in
+   *     can be transformed to localIdentity, if this principal id matches the principal id set in
+   *     "fs.azure.identity.transformer.service.principal.id" and localIdentity is stated in
    *     "fs.azure.identity.transformer.service.principal.substitution.list"
    *
    * 3. User principal name (UPN):
-   *     can be transformed to a short name(localUserOrGroup) if "fs.azure.identity.transformer.enable.short.name"
-   *     is enabled.
+   *     can be transformed to a short name(localIdentity) if originalIdentity is owner name, and
+   *     "fs.azure.identity.transformer.enable.short.name" is enabled.
    *
-   * @param originalUserOrGroup the original user or group in the get request results: FileStatus, AclStatus.
-   * @param localUserOrGroup the local user or group, should be parsed from UserGroupInformation.
+   * @param originalIdentity the original user or group in the get request results: FileStatus, AclStatus.
+   * @param isUserName indicate whether the input originalIdentity is an owner name or owning group name.
+   * @param localIdentity the local user or group, should be parsed from UserGroupInformation.
    * @return owner or group after transformation.
    * */
-  public String transformIdentityForGetRequest(String originalUserOrGroup, String localUserOrGroup) {
-    if (originalUserOrGroup == null) {
-      originalUserOrGroup = localUserOrGroup;
-      // localUserOrGroup might be a full name, so continue the transformation.
+  public String transformIdentityForGetRequest(String originalIdentity, boolean isUserName, String localIdentity) {
+    if (originalIdentity == null) {
+      originalIdentity = localIdentity;
+      // localIdentity might be a full name, so continue the transformation.
     }
     // case 1: it is $superuser and replace $superuser config is enabled
-    if (!skipSuperUserReplacement && SUPER_USER.equals(originalUserOrGroup)) {
-      return localUserOrGroup;
+    if (!skipSuperUserReplacement && SUPER_USER.equals(originalIdentity)) {
+      return localIdentity;
     }
 
     if (skipUserIdentityReplacement) {
-      return originalUserOrGroup;
+      return originalIdentity;
     }
 
     // case 2: original owner is principalId set in config, and localUser
     //         is a daemon service specified in substitution list,
     //         To avoid ownership check failure in job task, replace it
     //         to local daemon user/group
-    if (originalUserOrGroup.equals(servicePrincipalId) && isInSubstitutionList(localUserOrGroup)) {
-      return localUserOrGroup;
+    if (originalIdentity.equals(servicePrincipalId) && isInSubstitutionList(localIdentity)) {
+      return localIdentity;
     }
 
     // case 3: If original owner is a fully qualified name, and
     //         short name is enabled, replace with shortName.
-    if (shouldUseShortUserName(originalUserOrGroup)) {
-      return getShortName(originalUserOrGroup);
+    if (isUserName && shouldUseShortUserName(originalIdentity)) {
+      return getShortName(originalIdentity);
     }
 
-    return originalUserOrGroup;
+    return originalIdentity;
   }
 
   /**

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

@@ -153,13 +153,13 @@ public class ITestAbfsIdentityTransformer extends AbstractAbfsScaleTest{
     // with default config, identityTransformer should do $superUser replacement
     IdentityTransformer identityTransformer = getTransformerWithDefaultIdentityConfig(config);
     assertEquals("$superuser should be replaced with local user by default",
-            localUser, identityTransformer.transformIdentityForGetRequest(SUPER_USER, localUser));
+            localUser, identityTransformer.transformIdentityForGetRequest(SUPER_USER, true, localUser));
 
     // Disable $supeuser replacement
     config.setBoolean(FS_AZURE_SKIP_SUPER_USER_REPLACEMENT, true);
     identityTransformer = getTransformerWithCustomizedIdentityConfig(config);
     assertEquals("$superuser should not be replaced",
-            SUPER_USER, identityTransformer.transformIdentityForGetRequest(SUPER_USER, localUser));
+            SUPER_USER, identityTransformer.transformIdentityForGetRequest(SUPER_USER, true, localUser));
   }
 
   @Test
@@ -170,14 +170,14 @@ public class ITestAbfsIdentityTransformer extends AbstractAbfsScaleTest{
     // Default config
     IdentityTransformer identityTransformer = getTransformerWithDefaultIdentityConfig(config);
     assertEquals("By default servicePrincipalId should not be converted for GetFileStatus(), listFileStatus(), getAcl()",
-            SERVICE_PRINCIPAL_ID, identityTransformer.transformIdentityForGetRequest(SERVICE_PRINCIPAL_ID, localUser));
+            SERVICE_PRINCIPAL_ID, identityTransformer.transformIdentityForGetRequest(SERVICE_PRINCIPAL_ID, true, localUser));
 
     resetIdentityConfig(config);
     // 1. substitution list doesn't contain currentUser
     config.set(FS_AZURE_OVERRIDE_OWNER_SP_LIST, "a,b,c,d");
     identityTransformer = getTransformerWithCustomizedIdentityConfig(config);
     assertEquals("servicePrincipalId should not be replaced if local daemon user is not in substitution list",
-            SERVICE_PRINCIPAL_ID, identityTransformer.transformIdentityForGetRequest(SERVICE_PRINCIPAL_ID, localUser));
+            SERVICE_PRINCIPAL_ID, identityTransformer.transformIdentityForGetRequest(SERVICE_PRINCIPAL_ID, true, localUser));
 
     resetIdentityConfig(config);
     // 2. substitution list contains currentUser(daemon name) but the service principal id in config doesn't match
@@ -185,7 +185,7 @@ public class ITestAbfsIdentityTransformer extends AbstractAbfsScaleTest{
     config.set(FS_AZURE_OVERRIDE_OWNER_SP, UUID.randomUUID().toString());
     identityTransformer = getTransformerWithCustomizedIdentityConfig(config);
     assertEquals("servicePrincipalId should not be replaced if it is not equal to the SPN set in config",
-            SERVICE_PRINCIPAL_ID, identityTransformer.transformIdentityForGetRequest(SERVICE_PRINCIPAL_ID, localUser));
+            SERVICE_PRINCIPAL_ID, identityTransformer.transformIdentityForGetRequest(SERVICE_PRINCIPAL_ID, true, localUser));
 
     resetIdentityConfig(config);
     // 3. substitution list contains currentUser(daemon name) and the service principal id in config matches
@@ -193,7 +193,7 @@ public class ITestAbfsIdentityTransformer extends AbstractAbfsScaleTest{
     config.set(FS_AZURE_OVERRIDE_OWNER_SP, SERVICE_PRINCIPAL_ID);
     identityTransformer = getTransformerWithCustomizedIdentityConfig(config);
     assertEquals("servicePrincipalId should be transformed to local use",
-            localUser, identityTransformer.transformIdentityForGetRequest(SERVICE_PRINCIPAL_ID, localUser));
+            localUser, identityTransformer.transformIdentityForGetRequest(SERVICE_PRINCIPAL_ID, true, localUser));
 
     resetIdentityConfig(config);
     // 4. substitution is "*" but the service principal id in config doesn't match the input
@@ -201,7 +201,7 @@ public class ITestAbfsIdentityTransformer extends AbstractAbfsScaleTest{
     config.set(FS_AZURE_OVERRIDE_OWNER_SP, UUID.randomUUID().toString());
     identityTransformer = getTransformerWithCustomizedIdentityConfig(config);
     assertEquals("servicePrincipalId should not be replaced if it is not equal to the SPN set in config",
-            SERVICE_PRINCIPAL_ID, identityTransformer.transformIdentityForGetRequest(SERVICE_PRINCIPAL_ID, localUser));
+            SERVICE_PRINCIPAL_ID, identityTransformer.transformIdentityForGetRequest(SERVICE_PRINCIPAL_ID, true, localUser));
 
     resetIdentityConfig(config);
     // 5. substitution is "*" and the service principal id in config match the input
@@ -209,7 +209,7 @@ public class ITestAbfsIdentityTransformer extends AbstractAbfsScaleTest{
     config.set(FS_AZURE_OVERRIDE_OWNER_SP, SERVICE_PRINCIPAL_ID);
     identityTransformer = getTransformerWithCustomizedIdentityConfig(config);
     assertEquals("servicePrincipalId should be transformed to local user",
-            localUser, identityTransformer.transformIdentityForGetRequest(SERVICE_PRINCIPAL_ID, localUser));
+            localUser, identityTransformer.transformIdentityForGetRequest(SERVICE_PRINCIPAL_ID, true, localUser));
   }
 
   @Test
@@ -220,13 +220,16 @@ public class ITestAbfsIdentityTransformer extends AbstractAbfsScaleTest{
     // Default config
     IdentityTransformer identityTransformer = getTransformerWithDefaultIdentityConfig(config);
     assertEquals("full name should not be transformed if shortname is not enabled",
-            FULLY_QUALIFIED_NAME, identityTransformer.transformIdentityForGetRequest(FULLY_QUALIFIED_NAME, localUser));
+            FULLY_QUALIFIED_NAME, identityTransformer.transformIdentityForGetRequest(FULLY_QUALIFIED_NAME, true, localUser));
 
     // add config to get short name
     config.setBoolean(FS_AZURE_FILE_OWNER_ENABLE_SHORTNAME, true);
     identityTransformer = getTransformerWithCustomizedIdentityConfig(config);
-    assertEquals("should convert the full name to shortname ",
-            SHORT_NAME, identityTransformer.transformIdentityForGetRequest(FULLY_QUALIFIED_NAME, localUser));
+    assertEquals("should convert the full owner name to shortname ",
+            SHORT_NAME, identityTransformer.transformIdentityForGetRequest(FULLY_QUALIFIED_NAME, true, localUser));
+
+    assertEquals("group name should not be converted to shortname ",
+            FULLY_QUALIFIED_NAME, identityTransformer.transformIdentityForGetRequest(FULLY_QUALIFIED_NAME, false, localGroup));
   }
 
   @Test