HADOOP-15801. ABFS: Fixing skipUserGroupMetadata in AzureBlobFileSystemStore.

Contributed by Da Zhou

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

@@ -102,6 +102,8 @@ public class AzureBlobFileSystemStore {
   private AbfsClient client;
   private URI uri;
   private final UserGroupInformation userGroupInformation;
+  private final String userName;
+  private final String primaryUserGroup;
   private static final String DATE_TIME_PATTERN = "E, dd MMM yyyy HH:mm:ss 'GMT'";
   private static final String XMS_PROPERTIES_ENCODING = "ISO-8859-1";
   private static final int LIST_MAX_RESULTS = 5000;
@@ -128,6 +130,15 @@ public class AzureBlobFileSystemStore {
     }
 
     this.userGroupInformation = userGroupInformation;
+    this.userName = userGroupInformation.getShortUserName();
+
+    if (!abfsConfiguration.getSkipUserGroupMetadataDuringInitialization()) {
+      primaryUserGroup = userGroupInformation.getPrimaryGroupName();
+    } else {
+      //Provide a default group name
+      primaryUserGroup = userName;
+    }
+
     this.azureAtomicRenameDirSet = new HashSet<>(Arrays.asList(
         abfsConfiguration.getAzureAtomicRenameDirs().split(AbfsHttpConstants.COMMA)));
 
@@ -462,8 +473,8 @@ public class AzureBlobFileSystemStore {
       final boolean hasAcl = AbfsPermission.isExtendedAcl(permissions);
 
       return new VersionedFileStatus(
-              owner == null ? userGroupInformation.getUserName() : owner,
-              group == null ? userGroupInformation.getPrimaryGroupName() : group,
+              owner == null ? userName : owner,
+              group == null ? primaryUserGroup : group,
               permissions == null ? new AbfsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL)
                       : AbfsPermission.valueOf(permissions),
               hasAcl,
@@ -489,8 +500,8 @@ public class AzureBlobFileSystemStore {
       final boolean hasAcl = AbfsPermission.isExtendedAcl(permissions);
 
       return new VersionedFileStatus(
-              owner == null ? userGroupInformation.getUserName() : owner,
-              group == null ? userGroupInformation.getPrimaryGroupName() : group,
+              owner == null ? userName : owner,
+              group == null ? primaryUserGroup : group,
               permissions == null ? new AbfsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL)
                       : AbfsPermission.valueOf(permissions),
               hasAcl,
@@ -528,8 +539,8 @@ public class AzureBlobFileSystemStore {
       long blockSize = abfsConfiguration.getAzureBlockSize();
 
       for (ListResultEntrySchema entry : retrievedSchema.paths()) {
-        final String owner = entry.owner() == null ? userGroupInformation.getUserName() : entry.owner();
-        final String group = entry.group() == null ? userGroupInformation.getPrimaryGroupName() : entry.group();
+        final String owner = entry.owner() == null ? userName : entry.owner();
+        final String group = entry.group() == null ? primaryUserGroup : entry.group();
         final FsPermission fsPermission = entry.permissions() == null
                 ? new AbfsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL)
                 : AbfsPermission.valueOf(entry.permissions());
@@ -761,8 +772,8 @@ public class AzureBlobFileSystemStore {
             : AbfsPermission.valueOf(permissions);
 
     final AclStatus.Builder aclStatusBuilder = new AclStatus.Builder();
-    aclStatusBuilder.owner(owner == null ? userGroupInformation.getUserName() : owner);
-    aclStatusBuilder.group(group == null ? userGroupInformation.getPrimaryGroupName() : group);
+    aclStatusBuilder.owner(owner == null ? userName : owner);
+    aclStatusBuilder.group(group == null ? primaryUserGroup : group);
 
     aclStatusBuilder.setPermission(fsPermission);
     aclStatusBuilder.stickyBit(fsPermission.getStickyBit());