Support Mkdir/CreateFile with configured default umask in HDFS API

### What changes are proposed in this pull request?
Support creating directories and files with default permissions based on configuration propertykey in HDFS API.

### Why are the changes needed?
Alluxio Hdfs api hasn't the corresponding implementation of Mkdir without permission parameter. If it does not carry permission, the umask property in the configuration item shall prevail.

### Does this PR introduce any user facing changes?


			pr-link: #18253
			change-id: cid-6ddd2243bac00ebbdbdff1e731036c0d3d6228c8

core/client/hdfs/src/main/java/alluxio/hadoop/AbstractFileSystem.java

@@ -36,6 +36,7 @@
 import alluxio.master.MasterInquireClient.Factory;
 import alluxio.security.CurrentUser;
 import alluxio.security.authorization.Mode;
+import alluxio.util.ModeUtils;
 import alluxio.wire.BlockLocationInfo;
 import alluxio.wire.FileBlockInfo;
 import alluxio.wire.WorkerNetAddress;
@@ -150,6 +151,27 @@ public void close() throws IOException {
     mFileSystem.close();
   }
 
+  /**
+   * Attempts to create a file with default permission.
+   * Overwrite will not succeed if the path exists and is a folder.
+   *
+   * @param path path to create
+   * @param overwrite overwrite if file exists
+   * @param bufferSize the size in bytes of the buffer to be used
+   * @param replication under filesystem replication factor, this is ignored
+   * @param blockSize block size in bytes
+   * @param progress queryable progress
+   * @return an {@link FSDataOutputStream} created at the indicated path of a file
+   */
+  @Override
+  public FSDataOutputStream create(Path path, boolean overwrite, int bufferSize, short replication,
+                                   long blockSize, Progressable progress) throws IOException {
+    String confUmask = mAlluxioConf.getString(PropertyKey.SECURITY_AUTHORIZATION_PERMISSION_UMASK);
+    Mode mode = ModeUtils.applyFileUMask(Mode.defaults(), confUmask);
+    return this.create(path, new FsPermission(mode.toShort()), overwrite, bufferSize, replication,
+        blockSize, progress);
+  }
+
   /**
    * Attempts to create a file. Overwrite will not succeed if the path exists and is a folder.
    *
@@ -601,6 +623,20 @@ public FileStatus[] listStatus(Path path) throws IOException {
     return ret;
   }
 
+  /**
+   * Attempts to create a folder with the specified path with default permission.
+   * Parent directories will be created.
+   *
+   * @param path path to create
+   * @return true if the indicated folder is created successfully or already exists
+   */
+  @Override
+  public boolean mkdirs(Path path) throws IOException {
+    String confUmask = mAlluxioConf.getString(PropertyKey.SECURITY_AUTHORIZATION_PERMISSION_UMASK);
+    Mode mode = ModeUtils.applyDirectoryUMask(Mode.defaults(), confUmask);
+    return mkdirs(path, new FsPermission(mode.toShort()));
+  }
+
   /**
    * Attempts to create a folder with the specified path. Parent directories will be created.
    *

core/client/hdfs/src/test/java/alluxio/hadoop/AbstractFileSystemTest.java

@@ -52,6 +52,8 @@
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsCreateModes;
+import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.junit.After;
 import org.junit.Before;
@@ -726,7 +728,9 @@ public void createWithoutOverwrite() throws Exception {
             ExceptionMessage.CANNOT_OVERWRITE_FILE_WITHOUT_OVERWRITE.getMessage(path.toString())));
 
     try (FileSystem alluxioHadoopFs = new FileSystem(alluxioFs)) {
-      alluxioHadoopFs.create(path, false, 100, (short) 1, 1000);
+      alluxioHadoopFs.create(path,
+          FsCreateModes.applyUMask(FsPermission.getFileDefault(), FsPermission.getUMask(getConf())),
+          false, 100, (short) 1, 1000, null);
       fail("create() of existing file is expected to fail");
     } catch (IOException e) {
       assertEquals("alluxio.exception.FileAlreadyExistsException: "

tests/src/test/java/alluxio/client/hadoop/FileSystemAclIntegrationTest.java

@@ -108,6 +108,12 @@ public void cleanupTFS() throws Exception {
 
   @Test
   public void createFileWithPermission() throws Exception {
+    Path defaultFile = new Path("/createfile-default");
+    FSDataOutputStream stream = sTFS.create(defaultFile, false /* ignored */, 10 /* ignored */,
+        (short) 1 /* ignored */, 512 /* ignored */, null /* ignored */);
+    stream.close();
+    FileStatus fileStatus = sTFS.getFileStatus(defaultFile);
+    Assert.assertEquals((short) 0644, fileStatus.getPermission().toShort());
     List<Integer> permissionValues =
         Lists.newArrayList(0111, 0222, 0333, 0444, 0555, 0666, 0777, 0755, 0733, 0644, 0533, 0511);
     for (int value : permissionValues) {
@@ -124,6 +130,10 @@ public void createFileWithPermission() throws Exception {
 
   @Test
   public void mkdirsWithPermission() throws Exception {
+    Path defaultDir = new Path("/createDir-default");
+    sTFS.mkdirs(defaultDir);
+    FileStatus fileStatus = sTFS.getFileStatus(defaultDir);
+    Assert.assertEquals((short) 0755, fileStatus.getPermission().toShort());
     List<Integer> permissionValues =
         Lists.newArrayList(0111, 0222, 0333, 0444, 0555, 0666, 0777, 0755, 0733, 0644, 0533, 0511);
     for (int value : permissionValues) {