@Test public void testCreateTezSysStagingPath() throws Exception { String strAppId = "testAppId"; String expectedStageDir = RESOLVED_STAGE_DIR + File.separatorChar + TezCommonUtils.TEZ_SYSTEM_SUB_DIR + File.separatorChar + strAppId; String unResolvedStageDir = STAGE_DIR + File.separatorChar + TezCommonUtils.TEZ_SYSTEM_SUB_DIR + File.separatorChar + strAppId; Path stagePath = new Path(unResolvedStageDir); FileSystem fs = stagePath.getFileSystem(conf); if (fs.exists(stagePath)) { fs.delete(stagePath, true); } Assert.assertFalse(fs.exists(stagePath)); Path stageDir = TezCommonUtils.createTezSystemStagingPath(conf, strAppId); Assert.assertEquals(stageDir.toString(), expectedStageDir); Assert.assertTrue(fs.exists(stagePath)); }
@Private public static Path createTezSystemStagingPath(Configuration conf, String strAppId) { Path baseStagingPath = getTezBaseStagingPath(conf); Path tezStagingDir; try { tezStagingDir = new Path(baseStagingPath, TEZ_SYSTEM_SUB_DIR); FileSystem fs = tezStagingDir.getFileSystem(conf); tezStagingDir = new Path(tezStagingDir, strAppId); if (!fs.exists(tezStagingDir)) { mkDirForAM(fs, tezStagingDir); LOG.info("Tez system stage directory " + tezStagingDir + " doesn't exist and is created"); } } catch (IOException e) { throw new TezUncheckedException(e); } return tezStagingDir; }
TezCommonUtils { @Private public static Path createTezSystemStagingPath(Configuration conf, String strAppId) { Path baseStagingPath = getTezBaseStagingPath(conf); Path tezStagingDir; try { tezStagingDir = new Path(baseStagingPath, TEZ_SYSTEM_SUB_DIR); FileSystem fs = tezStagingDir.getFileSystem(conf); tezStagingDir = new Path(tezStagingDir, strAppId); if (!fs.exists(tezStagingDir)) { mkDirForAM(fs, tezStagingDir); LOG.info("Tez system stage directory " + tezStagingDir + " doesn't exist and is created"); } } catch (IOException e) { throw new TezUncheckedException(e); } return tezStagingDir; } }
TezCommonUtils { @Private public static Path createTezSystemStagingPath(Configuration conf, String strAppId) { Path baseStagingPath = getTezBaseStagingPath(conf); Path tezStagingDir; try { tezStagingDir = new Path(baseStagingPath, TEZ_SYSTEM_SUB_DIR); FileSystem fs = tezStagingDir.getFileSystem(conf); tezStagingDir = new Path(tezStagingDir, strAppId); if (!fs.exists(tezStagingDir)) { mkDirForAM(fs, tezStagingDir); LOG.info("Tez system stage directory " + tezStagingDir + " doesn't exist and is created"); } } catch (IOException e) { throw new TezUncheckedException(e); } return tezStagingDir; }  }
TezCommonUtils { @Private public static Path createTezSystemStagingPath(Configuration conf, String strAppId) { Path baseStagingPath = getTezBaseStagingPath(conf); Path tezStagingDir; try { tezStagingDir = new Path(baseStagingPath, TEZ_SYSTEM_SUB_DIR); FileSystem fs = tezStagingDir.getFileSystem(conf); tezStagingDir = new Path(tezStagingDir, strAppId); if (!fs.exists(tezStagingDir)) { mkDirForAM(fs, tezStagingDir); LOG.info("Tez system stage directory " + tezStagingDir + " doesn't exist and is created"); } } catch (IOException e) { throw new TezUncheckedException(e); } return tezStagingDir; }  static Path getTezBaseStagingPath(Configuration conf); @Private static Path createTezSystemStagingPath(Configuration conf, String strAppId); @Private static Path getTezSystemStagingPath(Configuration conf, String strAppId); @Private static Path getTezConfStagingPath(Path tezSysStagingPath); @Private static Path getTezSessionJarStagingPath(Path tezSysStagingPath); @Private static Path getTezBinPlanStagingPath(Path tezSysStagingPath); @Private static Path getTezTextPlanStagingPath(Path tezSysStagingPath); @Private static Path getRecoveryPath(Path tezSysStagingPath, Configuration conf); @Private static Path getAttemptRecoveryPath(Path recoveryPath, int attemptID); @Private static Path getDAGRecoveryPath(Path attemptRecoverPath, String dagID); @Private static Path getSummaryRecoveryPath(Path attemptRecoverPath); static void mkDirForAM(FileSystem fs, Path dir); static FSDataOutputStream createFileForAM(FileSystem fs, Path filePath); @Private static ByteString compressByteArrayToByteString(byte[] inBytes); @Private static byte[] decompressByteStringToByteArray(ByteString byteString); }
TezCommonUtils { @Private public static Path createTezSystemStagingPath(Configuration conf, String strAppId) { Path baseStagingPath = getTezBaseStagingPath(conf); Path tezStagingDir; try { tezStagingDir = new Path(baseStagingPath, TEZ_SYSTEM_SUB_DIR); FileSystem fs = tezStagingDir.getFileSystem(conf); tezStagingDir = new Path(tezStagingDir, strAppId); if (!fs.exists(tezStagingDir)) { mkDirForAM(fs, tezStagingDir); LOG.info("Tez system stage directory " + tezStagingDir + " doesn't exist and is created"); } } catch (IOException e) { throw new TezUncheckedException(e); } return tezStagingDir; }  static Path getTezBaseStagingPath(Configuration conf); @Private static Path createTezSystemStagingPath(Configuration conf, String strAppId); @Private static Path getTezSystemStagingPath(Configuration conf, String strAppId); @Private static Path getTezConfStagingPath(Path tezSysStagingPath); @Private static Path getTezSessionJarStagingPath(Path tezSysStagingPath); @Private static Path getTezBinPlanStagingPath(Path tezSysStagingPath); @Private static Path getTezTextPlanStagingPath(Path tezSysStagingPath); @Private static Path getRecoveryPath(Path tezSysStagingPath, Configuration conf); @Private static Path getAttemptRecoveryPath(Path recoveryPath, int attemptID); @Private static Path getDAGRecoveryPath(Path attemptRecoverPath, String dagID); @Private static Path getSummaryRecoveryPath(Path attemptRecoverPath); static void mkDirForAM(FileSystem fs, Path dir); static FSDataOutputStream createFileForAM(FileSystem fs, Path filePath); @Private static ByteString compressByteArrayToByteString(byte[] inBytes); @Private static byte[] decompressByteStringToByteArray(ByteString byteString); static final FsPermission TEZ_AM_DIR_PERMISSION; static final FsPermission TEZ_AM_FILE_PERMISSION; static final String TEZ_SYSTEM_SUB_DIR; }