class FileSystemAccessService extends BaseService implements FileSystemAccess {

  @Override
  protected void init() throws ServiceException {
    LOG.info("Using FileSystemAccess JARs version [{}]",
        VersionInfo.getVersion());
    String security =
        getServiceConfig().get(AUTHENTICATION_TYPE, "simple").trim();
    if (security.equals("kerberos")) {
      String defaultName = getServer().getName();
      String keytab =
          System.getProperty("user.home") + "/" + defaultName + ".keytab";
      keytab = getServiceConfig().get(KERBEROS_KEYTAB, keytab).trim();
      if (keytab.length() == 0) {
        throw new ServiceException(FileSystemAccessException.ERROR.H01,
            KERBEROS_KEYTAB);
      }
      String principal = defaultName + "/localhost@LOCALHOST";
      principal = getServiceConfig().get(KERBEROS_PRINCIPAL, principal).trim();
      if (principal.length() == 0) {
        throw new ServiceException(FileSystemAccessException.ERROR.H01,
            KERBEROS_PRINCIPAL);
      }
      Configuration conf = new Configuration();
      conf.set("hadoop.security.authentication", "kerberos");
      UserGroupInformation.setConfiguration(conf);
      try {
        UserGroupInformation.loginUserFromKeytab(principal, keytab);
      } catch (IOException ex) {
        throw new ServiceException(FileSystemAccessException.ERROR.H02,
            ex.getMessage(), ex);
      }
      LOG.info(
          "Using FileSystemAccess Kerberos authentication, principal [{}] keytab [{}]",
          principal, keytab);
    } else if (security.equals("simple")) {
      Configuration conf = new Configuration();
      conf.set("hadoop.security.authentication", "simple");
      UserGroupInformation.setConfiguration(conf);
      LOG.info(
          "Using FileSystemAccess simple/pseudo authentication, principal [{}]",
          System.getProperty("user.name"));
    } else {
      throw new ServiceException(FileSystemAccessException.ERROR.H09, security);
    }

    String hadoopConfDirProp =
        getServiceConfig().get(HADOOP_CONF_DIR, getServer().getConfigDir());
    File hadoopConfDir = new File(hadoopConfDirProp).getAbsoluteFile();
    if (!hadoopConfDir.exists()) {
      hadoopConfDir = new File(getServer().getConfigDir()).getAbsoluteFile();
    }
    if (!hadoopConfDir.exists()) {
      throw new ServiceException(FileSystemAccessException.ERROR.H10,
          hadoopConfDir);
    }
    try {
      serviceHadoopConf = loadHadoopConf(hadoopConfDir);
    } catch (IOException ex) {
      throw new ServiceException(FileSystemAccessException.ERROR.H11,
          ex.toString(), ex);
    }

    LOG.debug("FileSystemAccess FileSystem configuration:");
    for (Map.Entry entry : serviceHadoopConf) {
      LOG.debug("  {} = {}", entry.getKey(), entry.getValue());
    }
    setRequiredServiceHadoopConf(serviceHadoopConf);

    nameNodeWhitelist = toLowerCase(
        getServiceConfig().getTrimmedStringCollection(NAME_NODE_WHITELIST));
  }

  public  FileSystemAccessService();

}