diff --git a/modules/api/src/main/java/org/apache/fluo/api/config/FluoConfiguration.java b/modules/api/src/main/java/org/apache/fluo/api/config/FluoConfiguration.java
index cf416c0cb..466763fe8 100644
--- a/modules/api/src/main/java/org/apache/fluo/api/config/FluoConfiguration.java
+++ b/modules/api/src/main/java/org/apache/fluo/api/config/FluoConfiguration.java
@@ -48,6 +48,18 @@ public class FluoConfiguration extends SimpleConfiguration {
 
   // Client properties
   private static final String CLIENT_PREFIX = FLUO_PREFIX + ".client";
+  /**
+   * @since 1.3.0
+   */
+  public static final String CLIENT_HDFS_KERBEROS = CLIENT_PREFIX + ".hdfs.kerberos";
+  /**
+   * @since 1.3.0
+   */
+  public static final String CLIENT_HDFS_KERBEROS_REALM = CLIENT_PREFIX + ".hdfs.kerberos.realm";
+  /**
+   * @since 1.3.0
+   */
+  public static final String CLIENT_HDFS_KERBEROS_KEYTAB = CLIENT_PREFIX + ".hdfs.kerberos.keytab";
   /**
    * @deprecated since 1.2.0 replaced by fluo.connection.application.name
    */
diff --git a/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java b/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java
index 5210cc0d6..3a6373259 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/client/FluoAdminImpl.java
@@ -59,6 +59,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.NodeExistsException;
 import org.slf4j.Logger;
@@ -76,6 +77,28 @@ public class FluoAdminImpl implements FluoAdmin {
 
   private final String appRootDir;
 
+  /**
+   * Kerberos autentication method.
+   * 
+   * @param realm Realm to be used in authentication.
+   * @param keytab Keytab path.
+   * @since 1.3.0
+   */
+  public void hdfsLoginWithKerberos(final String realm, final String keytab) {
+
+    try {
+      Configuration conf = new Configuration();
+      conf.set("hadoop.security.authentication", "kerberos");
+      conf.set("hadoop.security.authorization", "true");
+      UserGroupInformation.setConfiguration(conf);
+      UserGroupInformation.loginUserFromKeytab(realm, keytab);
+
+      logger.info("Connected with REALM: '{}'.", realm);
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   public FluoAdminImpl(FluoConfiguration config) {
     this.config = config;
 
@@ -373,6 +396,14 @@ public static String copyDirToDfs(String dfsRoot, String appName, String srcDir,
   }
 
   private String copyJarsToDfs(String jars, String destDir) {
+
+    if (config.getClientConfiguration().getBoolean(FluoConfiguration.CLIENT_HDFS_KERBEROS, false)) {
+      this.hdfsLoginWithKerberos(
+          config.getClientConfiguration().getString(FluoConfiguration.CLIENT_HDFS_KERBEROS_REALM, ""),
+          config.getClientConfiguration().getString(FluoConfiguration.CLIENT_HDFS_KERBEROS_KEYTAB, ""));
+    }
+
+
     String dfsAppRoot = config.getDfsRoot() + "/" + config.getApplicationName();
     String dfsDestDir = dfsAppRoot + "/" + destDir;