StramClientUtils { public static Path getApexDFSRootDir(FileSystem fs, Configuration conf) { String apexDfsRootDir = conf.get(APEX_APP_DFS_ROOT_DIR); boolean useImpersonated = conf.getBoolean(StramUserLogin.DT_APP_PATH_IMPERSONATED, false); String userShortName = null; if (useImpersonated) { try { userShortName = UserGroupInformation.getCurrentUser().getShortUserName(); } catch (IOException ex) { LOG.warn("Error getting current/login user name {}", apexDfsRootDir, ex); } } if (!useImpersonated || userShortName == null) { return getDTDFSRootDir(fs, conf); } if (StringUtils.isBlank(apexDfsRootDir)) { apexDfsRootDir = getDefaultRootFolder(); } return evalDFSRootDir(fs, conf, apexDfsRootDir, userShortName, true); }  static YarnClient createYarnClient(Configuration conf); static String getHostName(); static File getUserDTDirectory(); static File getConfigDir(); static File getInstallationDir(); static boolean isDevelopmentMode(); static File getBackupsDirectory(); static Configuration addDTDefaultResources(Configuration conf); static Configuration addDTSiteResources(Configuration conf); static void addDTLocalResources(Configuration conf); static URL getDTSiteXmlFile(); static FileSystem newFileSystemInstance(Configuration conf); static Path getApexDFSRootDir(FileSystem fs, Configuration conf); static Path getDTDFSRootDir(FileSystem fs, Configuration conf); static Path getDTDFSConfigDir(FileSystem fs, Configuration conf); static Path getDTDFSProfilesDir(FileSystem fs, Configuration conf); static void changeDTEnvironment(String key, String value); static void copyFromLocalFileNoChecksum(FileSystem fs, File fromLocal, Path toDFS); static boolean configComplete(Configuration conf); static void evalProperties(Properties target, Configuration vars); static void evalConfiguration(Configuration conf); static T doAs(String userName, PrivilegedExceptionAction<T> action); static ApplicationReport getStartedAppInstanceByName(YarnClient clientRMService, String appName, String user, String excludeAppId); static InetSocketAddress getRMWebAddress(Configuration conf, String rmId); static InetSocketAddress getRMWebAddress(Configuration conf, boolean sslEnabled, String rmId); static String getSocketConnectString(InetSocketAddress socketAddress); static List<InetSocketAddress> getRMAddresses(Configuration conf); static List<ApplicationReport> cleanAppDirectories(YarnClient clientRMService, Configuration conf, FileSystem fs, long finishedBefore); static AppPackage.AppInfo jsonFileToAppInfo(File file, Configuration config); static void addAttributeToArgs(Attribute<String> attribute, Context context, List<CharSequence> vargs); static final String DT_VERSION; static final String DT_DFS_ROOT_DIR; static final String APEX_APP_DFS_ROOT_DIR; static final String DT_DFS_USER_NAME; static final String DT_CONFIG_STATUS; static final String SUBDIR_APPS; static final String SUBDIR_PROFILES; static final String SUBDIR_CONF; static final long RESOURCEMANAGER_CONNECT_MAX_WAIT_MS_OVERRIDE; static final String DT_HDFS_TOKEN_MAX_LIFE_TIME; static final String DT_HDFS_TOKEN_RENEW_INTERVAL; static final String HDFS_TOKEN_MAX_LIFE_TIME; static final String HDFS_TOKEN_RENEW_INTERVAL; static final String DT_RM_TOKEN_MAX_LIFE_TIME; static final String DT_RM_TOKEN_RENEW_INTERVAL; @Deprecated
  static final String KEY_TAB_FILE; static final String TOKEN_ANTICIPATORY_REFRESH_FACTOR; static final long DELEGATION_TOKEN_MAX_LIFETIME_DEFAULT; static final long DELEGATION_TOKEN_RENEW_INTERVAL_DEFAULT; static final String TOKEN_REFRESH_PRINCIPAL; static final String TOKEN_REFRESH_KEYTAB; static final String DT_DEFAULT_XML_FILE; static final String DT_SITE_XML_FILE; static final String DT_SITE_GLOBAL_XML_FILE; static final String DT_ENV_SH_FILE; static final String CUSTOM_ENV_SH_FILE; static final String BACKUPS_DIRECTORY; }
@Test public void getApexDFSRootDirLegacy() throws IOException { Configuration conf = new YarnConfiguration(new Configuration(false)); conf.set(StramClientUtils.DT_DFS_ROOT_DIR, "/a/b/c"); conf.setBoolean(StramUserLogin.DT_APP_PATH_IMPERSONATED, false); FileSystem fs = FileSystem.newInstance(conf); Path path = StramClientUtils.getApexDFSRootDir(fs, conf); Assert.assertEquals("file:/a/b/c", path.toString()); }
@Test public void getApexDFSRootDirAbsPath() throws IOException { Configuration conf = new YarnConfiguration(new Configuration(false)); conf.set(StramClientUtils.APEX_APP_DFS_ROOT_DIR, "/x/y/z"); conf.setBoolean(StramUserLogin.DT_APP_PATH_IMPERSONATED, false); FileSystem fs = FileSystem.newInstance(conf); UserGroupInformation testUser = UserGroupInformation.createUserForTesting("testUser1", new String[]{""}); UserGroupInformation.setLoginUser(testUser); Path path = StramClientUtils.getApexDFSRootDir(fs, conf); Assert.assertEquals(fs.getHomeDirectory() + "/datatorrent", path.toString()); }
@Test public void getApexDFSRootDirScheme() throws IOException { Configuration conf = new YarnConfiguration(new Configuration(false)); conf.set(StramClientUtils.APEX_APP_DFS_ROOT_DIR, "file:/p/q/r"); conf.setBoolean(StramUserLogin.DT_APP_PATH_IMPERSONATED, false); FileSystem fs = FileSystem.newInstance(conf); UserGroupInformation testUser = UserGroupInformation.createUserForTesting("testUser1", new String[]{""}); UserGroupInformation.setLoginUser(testUser); Path path = StramClientUtils.getApexDFSRootDir(fs, conf); Assert.assertEquals(fs.getHomeDirectory() + "/datatorrent", path.toString()); }