[Fix-16918][Task] Make task working directory to 775 (#16923)

diff --git a/docs/docs/en/guide/security/security.md b/docs/docs/en/guide/security/security.md
index 646243a..fe11755 100644
--- a/docs/docs/en/guide/security/security.md
+++ b/docs/docs/en/guide/security/security.md

@@ -19,7 +19,9 @@
 - Tenant Code: **The tenant code is the user on Linux, unique and cannot be repeated**
 - The administrator enters the `Security Center->Tenant Management` page, and clicks the `Create Tenant` button to create a tenant.
 
-> Note: Currently, only admin users can modify tenant.
+> Note:
+> 1. Currently, only admin users can modify tenant.
+> 2. If you create a tenant manually in the Linux, you need to add the manually created tenant to the dolphinscheduler bootstrap user's group, so that the tenant will have enough working directory permissions.
 
 ![create-tenant](../../../../img/new_ui/dev/security/create-tenant.png)
 

diff --git a/docs/docs/zh/guide/security/security.md b/docs/docs/zh/guide/security/security.md
index a48954c..43e90fb 100644
--- a/docs/docs/zh/guide/security/security.md
+++ b/docs/docs/zh/guide/security/security.md

@@ -18,7 +18,9 @@
 - 租户编码：**租户编码是 Linux上 的用户，唯一，不能重复**
 - 管理员进入安全中心->租户管理页面，点击“创建租户”按钮，创建租户。
 
-> 注意：目前仅有 admin 用户可以修改租户。
+> 注意：
+> 1. 目前仅有 admin 用户可以修改租户；
+> 2. 如果您在 Linux 中手动创建一个租户，则需要将手动创建的租户添加到 dolphinscheduler 启动用户组，以便该租户拥有足够的工作目录权限。
 
 ![create-tenant](../../../../img/new_ui/dev/security/create-tenant.png)
 

diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/FileUtils.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/FileUtils.java
index 685dc63..c826d71 100644
--- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/FileUtils.java
+++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/FileUtils.java

@@ -62,7 +62,9 @@
 
     public static final String KUBE_CONFIG_FILE = "config";
 
-    private static final Set<PosixFilePermission> PERMISSION_755 = PosixFilePermissions.fromString("rwxr-xr-x");
+    public static final Set<PosixFilePermission> PERMISSION_755 = PosixFilePermissions.fromString("rwxr-xr-x");
+
+    public static final Set<PosixFilePermission> PERMISSION_775 = PosixFilePermissions.fromString("rwxrwxr-x");
 
     /**
      * get download file absolute path and name
@@ -239,7 +241,7 @@
     public static void createFileWith755(@NonNull Path path) throws IOException {
         final Path parent = path.getParent();
         if (!parent.toFile().exists()) {
-            createDirectoryWith755(parent);
+            createDirectoryWithPermission(parent, PERMISSION_755);
         }
         if (SystemUtils.IS_OS_WINDOWS) {
             Files.createFile(path);
@@ -249,29 +251,6 @@
         }
     }
 
-    public static void createDirectoryWith755(@NonNull Path path) throws IOException {
-        if (path.toFile().exists()) {
-            return;
-        }
-        if (OSUtils.isWindows()) {
-            Files.createDirectories(path);
-        } else {
-            Path parent = path.getParent();
-            if (parent != null && !parent.toFile().exists()) {
-                createDirectoryWith755(parent);
-            }
-
-            try {
-                Files.createDirectory(path);
-                Files.setPosixFilePermissions(path, PERMISSION_755);
-            } catch (FileAlreadyExistsException fileAlreadyExistsException) {
-                // Catch the FileAlreadyExistsException here to avoid create the same parent directory in parallel
-                log.debug("The directory: {} already exists", path);
-            }
-
-        }
-    }
-
     public static void setFileTo755(File file) throws IOException {
         if (OSUtils.isWindows()) {
             return;
@@ -289,6 +268,29 @@
         }
     }
 
+    public static void createDirectoryWithPermission(@NonNull Path path,
+                                                     @NonNull Set<PosixFilePermission> permissions) throws IOException {
+        if (path.toFile().exists()) {
+            return;
+        }
+
+        if (OSUtils.isWindows()) {
+            Files.createDirectories(path);
+        } else {
+            Path parent = path.getParent();
+            if (parent != null && !parent.toFile().exists()) {
+                createDirectoryWithPermission(parent, permissions);
+            }
+
+            try {
+                Files.createDirectory(path);
+                Files.setPosixFilePermissions(path, permissions);
+            } catch (FileAlreadyExistsException fileAlreadyExistsException) {
+                log.error("The directory: {} already exists", path);
+            }
+        }
+    }
+
     public static String concatFilePath(String... paths) {
         if (paths.length == 0) {
             throw new IllegalArgumentException("At least one path should be provided");

diff --git a/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/FileUtilsTest.java b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/FileUtilsTest.java
index 581ee8e..2396d52 100644
--- a/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/FileUtilsTest.java
+++ b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/FileUtilsTest.java

@@ -60,10 +60,10 @@
     }
 
     @Test
-    public void createDirectoryWith755() throws IOException {
+    public void testCreateDirectoryWithPermission() throws IOException {
         Path path = Paths.get("/tmp/createWorkDirAndUserIfAbsent");
         try {
-            FileUtils.createDirectoryWith755(path);
+            FileUtils.createDirectoryWithPermission(path, FileUtils.PERMISSION_755);
             File file = path.toFile();
             Assertions.assertTrue(file.exists());
             Assertions.assertTrue(file.isDirectory());
@@ -71,7 +71,7 @@
             Assertions.assertTrue(file.canRead());
             Assertions.assertTrue(file.canWrite());
 
-            FileUtils.createDirectoryWith755(Paths.get("/"));
+            FileUtils.createDirectoryWithPermission(Paths.get("/"), FileUtils.PERMISSION_755);
         } catch (Exception e) {
             e.printStackTrace();
             Assertions.fail(e.getMessage());

diff --git a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-abs/src/main/java/org/apache/dolphinscheduler/plugin/storage/abs/AbsStorageOperator.java b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-abs/src/main/java/org/apache/dolphinscheduler/plugin/storage/abs/AbsStorageOperator.java
index 0b60300..edd2b86 100644
--- a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-abs/src/main/java/org/apache/dolphinscheduler/plugin/storage/abs/AbsStorageOperator.java
+++ b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-abs/src/main/java/org/apache/dolphinscheduler/plugin/storage/abs/AbsStorageOperator.java

@@ -99,7 +99,7 @@
         if (dstFile.isDirectory()) {
             Files.delete(dstFile.toPath());
         } else {
-            FileUtils.createDirectoryWith755(dstFile.getParentFile().toPath());
+            FileUtils.createDirectoryWithPermission(dstFile.getParentFile().toPath(), FileUtils.PERMISSION_755);
         }
 
         BlobClient blobClient = blobContainerClient.getBlobClient(srcFilePath);

diff --git a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-cos/src/main/java/org/apache/dolphinscheduler/plugin/storage/cos/CosStorageOperator.java b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-cos/src/main/java/org/apache/dolphinscheduler/plugin/storage/cos/CosStorageOperator.java
index e2f605e..c4d1468 100644
--- a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-cos/src/main/java/org/apache/dolphinscheduler/plugin/storage/cos/CosStorageOperator.java
+++ b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-cos/src/main/java/org/apache/dolphinscheduler/plugin/storage/cos/CosStorageOperator.java

@@ -148,7 +148,7 @@
         if (dstFile.isDirectory()) {
             Files.delete(dstFile.toPath());
         } else {
-            FileUtils.createDirectoryWith755(dstFile.getParentFile().toPath());
+            FileUtils.createDirectoryWithPermission(dstFile.getParentFile().toPath(), FileUtils.PERMISSION_755);
         }
 
         GetObjectRequest getObjectRequest = new GetObjectRequest(bucketName, cosKey);

diff --git a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-gcs/src/main/java/org/apache/dolphinscheduler/plugin/storage/gcs/GcsStorageOperator.java b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-gcs/src/main/java/org/apache/dolphinscheduler/plugin/storage/gcs/GcsStorageOperator.java
index fe85a40..9036fb1 100644
--- a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-gcs/src/main/java/org/apache/dolphinscheduler/plugin/storage/gcs/GcsStorageOperator.java
+++ b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-gcs/src/main/java/org/apache/dolphinscheduler/plugin/storage/gcs/GcsStorageOperator.java

@@ -112,7 +112,7 @@
         if (dstFile.isDirectory()) {
             Files.delete(dstFile.toPath());
         } else {
-            FileUtils.createDirectoryWith755(dstFile.getParentFile().toPath());
+            FileUtils.createDirectoryWithPermission(dstFile.getParentFile().toPath(), FileUtils.PERMISSION_755);
         }
 
         Blob blob = gcsStorage.get(BlobId.of(bucketName, srcFilePath));

diff --git a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-obs/src/main/java/org/apache/dolphinscheduler/plugin/storage/obs/ObsStorageOperator.java b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-obs/src/main/java/org/apache/dolphinscheduler/plugin/storage/obs/ObsStorageOperator.java
index 129b9a8..fd96ff1 100644
--- a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-obs/src/main/java/org/apache/dolphinscheduler/plugin/storage/obs/ObsStorageOperator.java
+++ b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-obs/src/main/java/org/apache/dolphinscheduler/plugin/storage/obs/ObsStorageOperator.java

@@ -111,7 +111,7 @@
         if (dstFile.isDirectory()) {
             Files.delete(dstFile.toPath());
         } else {
-            FileUtils.createDirectoryWith755(dstFile.getParentFile().toPath());
+            FileUtils.createDirectoryWithPermission(dstFile.getParentFile().toPath(), FileUtils.PERMISSION_755);
         }
         ObsObject obsObject = obsClient.getObject(bucketName, srcFilePath);
         try (

diff --git a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-oss/src/main/java/org/apache/dolphinscheduler/plugin/storage/oss/OssStorageOperator.java b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-oss/src/main/java/org/apache/dolphinscheduler/plugin/storage/oss/OssStorageOperator.java
index 0b78ce4..c243c8a 100644
--- a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-oss/src/main/java/org/apache/dolphinscheduler/plugin/storage/oss/OssStorageOperator.java
+++ b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-oss/src/main/java/org/apache/dolphinscheduler/plugin/storage/oss/OssStorageOperator.java

@@ -169,7 +169,7 @@
         if (dstFile.isDirectory()) {
             Files.delete(dstFile.toPath());
         } else {
-            FileUtils.createDirectoryWith755(dstFile.getParentFile().toPath());
+            FileUtils.createDirectoryWithPermission(dstFile.getParentFile().toPath(), FileUtils.PERMISSION_755);
         }
         OSSObject ossObject = ossClient.getObject(bucketName, srcFilePath);
         try (

diff --git a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/main/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperator.java b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/main/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperator.java
index d1836d7..2f4954f 100644
--- a/dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/main/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperator.java
+++ b/dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/main/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperator.java

@@ -113,7 +113,7 @@
         if (dstFile.isDirectory()) {
             Files.delete(dstFile.toPath());
         } else {
-            FileUtils.createDirectoryWith755(dstFile.getParentFile().toPath());
+            FileUtils.createDirectoryWithPermission(dstFile.getParentFile().toPath(), FileUtils.PERMISSION_755);
         }
         S3Object o = s3Client.getObject(bucketName, srcFilePath);
         try (

diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionContextUtils.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionContextUtils.java
index cbe2a6f..79536bb 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionContextUtils.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionContextUtils.java

@@ -50,7 +50,8 @@
                 log.warn("The TaskInstance WorkingDirectory: {} is exist, will recreate again",
                         taskInstanceWorkingDirectory);
             }
-            FileUtils.createDirectoryWith755(Paths.get(taskInstanceWorkingDirectory));
+
+            FileUtils.createDirectoryWithPermission(Paths.get(taskInstanceWorkingDirectory), FileUtils.PERMISSION_775);
 
             taskExecutionContext.setExecutePath(taskInstanceWorkingDirectory);
             taskExecutionContext.setAppInfoPath(FileUtils.getAppInfoPath(taskInstanceWorkingDirectory));

diff --git a/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionContextUtilsTest.java b/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionContextUtilsTest.java
index 328006f..05d4196 100644
--- a/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionContextUtilsTest.java
+++ b/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionContextUtilsTest.java

@@ -44,7 +44,7 @@
         try {
             // Test if the working directory is exist
             // will delete it and recreate
-            FileUtils.createDirectoryWith755(Paths.get(taskWorkingDirectory));
+            FileUtils.createDirectoryWithPermission(Paths.get(taskWorkingDirectory), FileUtils.PERMISSION_775);
             Files.createFile(Paths.get(taskWorkingDirectory, "text.txt"));
             Assertions.assertTrue(Files.exists(Paths.get(taskWorkingDirectory, "text.txt")));