@Override
  public void load(String mountTableConfigPath, Configuration conf)
      throws IOException {
    this.mountTable = new Path(mountTableConfigPath);
    String scheme = mountTable.toUri().getScheme();
    FsGetter fsGetter = new ViewFileSystemOverloadScheme.ChildFsGetter(scheme);
    try (FileSystem fs = fsGetter.getNewInstance(mountTable.toUri(), conf)) {
      RemoteIterator<LocatedFileStatus> listFiles =
          fs.listFiles(mountTable, false);
      LocatedFileStatus lfs = null;
      int higherVersion = -1;
      while (listFiles.hasNext()) {
        LocatedFileStatus curLfs = listFiles.next();
        String cur = curLfs.getPath().getName();
        String[] nameParts = cur.split(REGEX_DOT);
        if (nameParts.length < 2) {
          logInvalidFileNameFormat(cur);
          continue; // invalid file name
        }
        int curVersion = higherVersion;
        try {
          curVersion = Integer.parseInt(nameParts[nameParts.length - 2]);
        } catch (NumberFormatException nfe) {
          logInvalidFileNameFormat(cur);
          continue;
        }

        if (curVersion > higherVersion) {
          higherVersion = curVersion;
          lfs = curLfs;
        }
      }

      if (lfs == null) {
        // No valid mount table file found.
        // TODO: Should we fail? Currently viewfs init will fail if no mount
        // links anyway.
        LOGGER.warn("No valid mount-table file exist at: {}. At least one "
            + "mount-table file should present with the name format: "
            + "mount-table.<versionNumber>.xml", mountTableConfigPath);
        return;
      }
      // Latest version file.
      Path latestVersionMountTable = lfs.getPath();
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Loading the mount-table {} into configuration.",
            latestVersionMountTable);
      }
      try (FSDataInputStream open = fs.open(latestVersionMountTable)) {
        Configuration newConf = new Configuration(false);
        newConf.addResource(open);
        // This will add configuration props as resource, instead of stream
        // itself. So, that stream can be closed now.
        conf.addResource(newConf);
      }
    }
  }
@Test
  public void testMountTableFileWithInvalidFormatWithNoDotsInName()
      throws Exception {
    Path path = new Path(new URI(targetTestRoot.toString()
        + "/testMountTableFileWithInvalidFormatWithNoDots/"));
    fsTarget.mkdirs(path);
    File invalidMountFileName =
        new File(new URI(path.toString() + "/tableInvalidVersionxml"));
    invalidMountFileName.createNewFile();
    // Pass mount table directory
    loader.load(path.toString(), conf);
    Assert.assertEquals(null, conf.get(MOUNT_LINK_KEY_SRC_TWO));
    Assert.assertEquals(null, conf.get(MOUNT_LINK_KEY_SRC_ONE));
    invalidMountFileName.delete();
  }