diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java
index 5c27692eb53..8235e937bdd 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java
@@ -125,4 +125,11 @@ public interface Constants {
       "fs.viewfs.ignore.port.in.mount.table.name";
 
   boolean CONFIG_VIEWFS_IGNORE_PORT_IN_MOUNT_TABLE_NAME_DEFAULT = false;
+
+  String CONFIG_VIEWFS_MOUNTTABLE_LOADER_IMPL =
+      CONFIG_VIEWFS_PREFIX + ".config.loader.impl";
+
+  Class<? extends MountTableConfigLoader>
+      DEFAULT_MOUNT_TABLE_CONFIG_LOADER_IMPL =
+      HCFSMountTableConfigLoader.class;
 }
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystemOverloadScheme.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystemOverloadScheme.java
index 12877ccf4bb..773793ba3b0 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystemOverloadScheme.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystemOverloadScheme.java
@@ -17,6 +17,10 @@
  */
 package org.apache.hadoop.fs.viewfs;
 
+import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_IGNORE_PORT_IN_MOUNT_TABLE_NAME;
+import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_MOUNTTABLE_LOADER_IMPL;
+import static org.apache.hadoop.fs.viewfs.Constants.DEFAULT_MOUNT_TABLE_CONFIG_LOADER_IMPL;
+
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.lang.reflect.Constructor;
@@ -30,8 +34,7 @@ import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FsConstants;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.UnsupportedFileSystemException;
-
-import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_IGNORE_PORT_IN_MOUNT_TABLE_NAME;
+import org.apache.hadoop.util.ReflectionUtils;
 
 /**
  * <p> This class is extended from the ViewFileSystem for the overloaded
@@ -160,7 +163,7 @@ public class ViewFileSystemOverloadScheme extends ViewFileSystem {
         conf.getBoolean(Constants.CONFIG_VIEWFS_IGNORE_PORT_IN_MOUNT_TABLE_NAME,
             true));
     if (null != mountTableConfigPath) {
-      MountTableConfigLoader loader = new HCFSMountTableConfigLoader();
+      MountTableConfigLoader loader = getMountTableConfigLoader(conf);
       loader.load(mountTableConfigPath, conf);
     } else {
       // TODO: Should we fail here.?
@@ -173,6 +176,30 @@ public class ViewFileSystemOverloadScheme extends ViewFileSystem {
     super.initialize(theUri, conf);
   }
 
+  private MountTableConfigLoader getMountTableConfigLoader(
+      final Configuration conf) {
+    Class<? extends MountTableConfigLoader> clazz =
+        conf.getClass(CONFIG_VIEWFS_MOUNTTABLE_LOADER_IMPL,
+            DEFAULT_MOUNT_TABLE_CONFIG_LOADER_IMPL,
+                MountTableConfigLoader.class);
+
+    if (clazz == null) {
+      throw new RuntimeException(
+          String.format("Errors on getting mount table loader class. "
+              + "The fs.viewfs.mounttable.config.loader.impl conf is %s. ",
+                  conf.get(CONFIG_VIEWFS_MOUNTTABLE_LOADER_IMPL,
+                      DEFAULT_MOUNT_TABLE_CONFIG_LOADER_IMPL.getName())));
+    }
+
+    try {
+      MountTableConfigLoader mountTableConfigLoader =
+          ReflectionUtils.newInstance(clazz, conf);
+      return mountTableConfigLoader;
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   /**
    * This method is overridden because in ViewFileSystemOverloadScheme if
    * overloaded scheme matches with mounted target fs scheme, file system