diff --git a/fs_brokers/apache_hdfs_broker/src/main/java/org/apache/doris/broker/hdfs/FileSystemManager.java b/fs_brokers/apache_hdfs_broker/src/main/java/org/apache/doris/broker/hdfs/FileSystemManager.java
index e80134078953bd..1d377b1c41a643 100644
--- a/fs_brokers/apache_hdfs_broker/src/main/java/org/apache/doris/broker/hdfs/FileSystemManager.java
+++ b/fs_brokers/apache_hdfs_broker/src/main/java/org/apache/doris/broker/hdfs/FileSystemManager.java
@@ -46,6 +46,7 @@
 import java.nio.ByteBuffer;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
+import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
 import java.util.Base64;
 import java.util.List;
@@ -65,7 +66,6 @@ public class FileSystemManager {
     private static final String HDFS_SCHEME = "hdfs";
     private static final String S3A_SCHEME = "s3a";
 
-    private static final String HDFS_UGI_CONF = "hadoop.job.ugi";
     private static final String USER_NAME_KEY = "username";
     private static final String PASSWORD_KEY = "password";
     private static final String AUTHENTICATION_SIMPLE = "simple";
@@ -246,9 +246,7 @@ public BrokerFileSystem getDistributedFileSystem(String path, Map<String, String
                 // TODO get this param from properties
                 // conf.set("dfs.replication", "2");
                 String tmpFilePath = null;
-                if (authentication.equals(AUTHENTICATION_SIMPLE)) {
-                    conf.set(HDFS_UGI_CONF, hdfsUgi);
-                } else if (authentication.equals(AUTHENTICATION_KERBEROS)){
+                if (authentication.equals(AUTHENTICATION_KERBEROS)){
                     conf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION,
                             AUTHENTICATION_KERBEROS);
 
@@ -287,9 +285,6 @@ public BrokerFileSystem getDistributedFileSystem(String path, Map<String, String
                                     e.getMessage());
                         }
                     }
-                } else {
-                    throw  new BrokerException(TBrokerOperationStatusCode.INVALID_ARGUMENT,
-                            "invalid authentication.");
                 }
                 if (!Strings.isNullOrEmpty(dfsNameServices)) {
                     // ha hdfs arguments
@@ -338,7 +333,20 @@ public BrokerFileSystem getDistributedFileSystem(String path, Map<String, String
                 }
 
                 conf.set(FS_HDFS_IMPL_DISABLE_CACHE, "true");
-                FileSystem dfsFileSystem = FileSystem.get(pathUri.getUri(), conf);
+                FileSystem dfsFileSystem = null;
+                if (authentication.equals(AUTHENTICATION_SIMPLE) &&
+                    properties.containsKey(USER_NAME_KEY)) {
+                    // Use the specified 'username' as the login name
+                    UserGroupInformation ugi = UserGroupInformation.createRemoteUser(username);
+                    dfsFileSystem = ugi.doAs(new PrivilegedExceptionAction<FileSystem>() {
+                        @Override
+                        public FileSystem run() throws Exception {
+                            return FileSystem.get(pathUri.getUri(), conf);
+                        }
+                    });
+                } else {
+                    dfsFileSystem = FileSystem.get(pathUri.getUri(), conf);
+                }
                 fileSystem.setFileSystem(dfsFileSystem);
             }
             return fileSystem;