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 463071a8b0..513da9b48e 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
@@ -25,6 +25,8 @@ import static org.apache.dolphinscheduler.common.constants.Constants.RESOURCE_VI
 import static org.apache.dolphinscheduler.common.constants.Constants.UTF_8;
 import static org.apache.dolphinscheduler.common.constants.DateConstants.YYYYMMDDHHMMSS;
 
+import org.apache.dolphinscheduler.common.constants.TenantConstants;
+
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang3.SystemUtils;
 
@@ -35,12 +37,15 @@ import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystems;
 import java.nio.file.Files;
 import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
 import java.nio.file.attribute.FileAttribute;
 import java.nio.file.attribute.PosixFilePermission;
 import java.nio.file.attribute.PosixFilePermissions;
+import java.nio.file.attribute.UserPrincipal;
+import java.nio.file.attribute.UserPrincipalLookupService;
 import java.util.Set;
 import java.util.zip.CRC32;
 import java.util.zip.CheckedInputStream;
@@ -323,6 +328,29 @@ public class FileUtils {
         return crcString;
     }
 
+    public static void setFileOwner(Path path, String tenant) {
+        try {
+            if (TenantConstants.DEFAULT_TENANT_CODE.equals(tenant)) {
+                log.debug("The current tenant: {} is the default tenant, no need to set the owner for file: {}", tenant,
+                        path);
+                return;
+            }
+            UserPrincipalLookupService userPrincipalLookupService =
+                    FileSystems.getDefault().getUserPrincipalLookupService();
+            UserPrincipal tenantPrincipal = userPrincipalLookupService.lookupPrincipalByName(tenant);
+            Files.setOwner(path, tenantPrincipal);
+        } catch (IOException e) {
+            log.error("Set file: {} owner to: {} failed", path, tenant, e);
+        }
+    }
+
+    public static void createDirectoryIfNotPresent(Path path) throws IOException {
+        if (Files.exists(path)) {
+            return;
+        }
+        Files.createDirectories(path);
+    }
+
     /**
      * Create a file with '755'.
      */
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionCheckerUtils.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionCheckerUtils.java
index c65f376039..a8beaa8356 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionCheckerUtils.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionCheckerUtils.java
@@ -34,13 +34,9 @@ import org.apache.commons.lang3.SystemUtils;
 import org.apache.commons.lang3.tuple.Pair;
 
 import java.io.File;
-import java.io.IOException;
-import java.nio.file.FileSystems;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.nio.file.attribute.UserPrincipal;
-import java.nio.file.attribute.UserPrincipalLookupService;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -97,9 +93,9 @@ public class TaskExecutionCheckerUtils {
             taskExecutionContext.setExecutePath(execLocalPath);
             taskExecutionContext.setAppInfoPath(FileUtils.getAppInfoPath(execLocalPath));
             Path executePath = Paths.get(taskExecutionContext.getExecutePath());
-            createDirectory(executePath);
-            if (!TenantConstants.DEFAULT_TENANT_CODE.equals(taskExecutionContext.getTenantCode())) {
-                setOwner(executePath, taskExecutionContext.getTenantCode());
+            FileUtils.createDirectoryIfNotPresent(executePath);
+            if (OSUtils.isSudoEnable()) {
+                FileUtils.setFileOwner(executePath, taskExecutionContext.getTenantCode());
             }
         } catch (Throwable ex) {
             throw new TaskException("Cannot create process execute dir", ex);
@@ -126,7 +122,7 @@ public class TaskExecutionCheckerUtils {
             if (notExist) {
                 downloadFiles.add(Pair.of(fullName, fileName));
             } else {
-                log.info("file : {} exists ", resFile.getName());
+                log.warn("Resource file : {} already exists will not download again ", resFile.getName());
             }
         });
         if (!downloadFiles.isEmpty() && !PropertyUtils.isResourceStorageStartup()) {
@@ -141,8 +137,11 @@ public class TaskExecutionCheckerUtils {
                     log.info("get resource file from path:{}", fullName);
 
                     long resourceDownloadStartTime = System.currentTimeMillis();
-                    storageOperate.download(actualTenant, fullName,
-                            execLocalPath + File.separator + fileName, true);
+                    storageOperate.download(actualTenant, fullName, execLocalPath + File.separator + fileName, true);
+                    if (OSUtils.isSudoEnable()) {
+                        FileUtils.setFileOwner(Paths.get(execLocalPath, fileName),
+                                taskExecutionContext.getTenantCode());
+                    }
                     WorkerServerMetrics
                             .recordWorkerResourceDownloadTime(System.currentTimeMillis() - resourceDownloadStartTime);
                     WorkerServerMetrics.recordWorkerResourceDownloadSize(
@@ -156,29 +155,4 @@ public class TaskExecutionCheckerUtils {
         }
     }
 
-    private static void createDirectory(Path filePath) {
-        if (Files.exists(filePath)) {
-            return;
-        }
-        try {
-            Files.createDirectories(filePath);
-        } catch (IOException e) {
-            throw new TaskException("Create directory " + filePath + " failed ", e);
-        }
-    }
-
-    private static void setOwner(Path filePath, String tenant) {
-        try {
-            if (!OSUtils.isSudoEnable()) {
-                // we need to open sudo, then we can change the owner.
-                return;
-            }
-            UserPrincipalLookupService userPrincipalLookupService =
-                    FileSystems.getDefault().getUserPrincipalLookupService();
-            UserPrincipal tenantPrincipal = userPrincipalLookupService.lookupPrincipalByName(tenant);
-            Files.setOwner(filePath, tenantPrincipal);
-        } catch (IOException e) {
-            throw new TaskException("Set tenant directory " + filePath + " permission failed, tenant: " + tenant, e);
-        }
-    }
 }