@Test public void testSafeRenameRecursively() throws Exception { final Logger log = LoggerFactory.getLogger("HadoopUtilsTest.testSafeRenameRecursively"); final Path hadoopUtilsTestDir = new Path(Files.createTempDir().getAbsolutePath(), "HadoopUtilsTestDir"); final FileSystem fs = FileSystem.getLocal(new Configuration()); try { for (int i = 0; i < 100; i++) { fs.mkdirs(hadoopUtilsTestDir); fs.mkdirs(new Path(hadoopUtilsTestDir, "testSafeRename/a/b/c")); fs.mkdirs(new Path(hadoopUtilsTestDir, "testRenameStaging1/a/b/c")); fs.mkdirs(new Path(hadoopUtilsTestDir, "testRenameStaging1/a/b/c/e")); fs.create(new Path(hadoopUtilsTestDir, "testRenameStaging1/a/b/c/t1.txt")); fs.create(new Path(hadoopUtilsTestDir, "testRenameStaging1/a/b/c/e/t2.txt")); fs.mkdirs(new Path(hadoopUtilsTestDir, "testRenameStaging2/a/b/c")); fs.mkdirs(new Path(hadoopUtilsTestDir, "testRenameStaging2/a/b/c/e")); fs.create(new Path(hadoopUtilsTestDir, "testRenameStaging2/a/b/c/t3.txt")); fs.create(new Path(hadoopUtilsTestDir, "testRenameStaging2/a/b/c/e/t4.txt")); ExecutorService executorService = Executors.newFixedThreadPool(2); final Throwable[] runnableErrors = {null, null}; Future<?> renameFuture = executorService.submit(new Runnable() { @Override public void run() { try { HadoopUtils.renameRecursively(fs, new Path(hadoopUtilsTestDir, "testRenameStaging1"), new Path( hadoopUtilsTestDir, "testSafeRename")); } catch (Throwable e) { log.error("Rename error: " + e, e); runnableErrors[0] = e; } } }); Future<?> safeRenameFuture = executorService.submit(new Runnable() { @Override public void run() { try { HadoopUtils.safeRenameRecursively(fs, new Path(hadoopUtilsTestDir, "testRenameStaging2"), new Path( hadoopUtilsTestDir, "testSafeRename")); } catch (Throwable e) { log.error("Safe rename error: " + e, e); runnableErrors[1] = e; } } }); renameFuture.get(10, TimeUnit.SECONDS); safeRenameFuture.get(10, TimeUnit.SECONDS); executorService.shutdownNow(); Assert.assertNull(runnableErrors[0], "Runnable 0 error: " + runnableErrors[0]); Assert.assertNull(runnableErrors[1], "Runnable 1 error: " + runnableErrors[1]); Assert.assertTrue(fs.exists(new Path(hadoopUtilsTestDir, "testSafeRename/a/b/c/t1.txt"))); Assert.assertTrue(fs.exists(new Path(hadoopUtilsTestDir, "testSafeRename/a/b/c/t3.txt"))); Assert.assertTrue(!fs.exists(new Path(hadoopUtilsTestDir, "testSafeRename/a/b/c/e/e/t2.txt"))); Assert.assertTrue(fs.exists(new Path(hadoopUtilsTestDir, "testSafeRename/a/b/c/e/t2.txt"))); Assert.assertTrue(fs.exists(new Path(hadoopUtilsTestDir, "testSafeRename/a/b/c/e/t4.txt"))); fs.delete(hadoopUtilsTestDir, true); } } finally { fs.delete(hadoopUtilsTestDir, true); } }
public static void safeRenameRecursively(FileSystem fileSystem, Path from, Path to) throws IOException { for (FileStatus fromFile : FileListUtils.listFilesRecursively(fileSystem, from)) { Path relativeFilePath = new Path(StringUtils.substringAfter(fromFile.getPath().toString(), from.toString() + Path.SEPARATOR)); Path toFilePath = new Path(to, relativeFilePath); if (!fileSystem.exists(toFilePath)) { boolean renamed = false; for (int i = 0; !renamed && i < MAX_RENAME_TRIES; i++) { try { renamed = fileSystem.rename(fromFile.getPath(), toFilePath); break; } catch (FileNotFoundException e) { if (i + 1 >= MAX_RENAME_TRIES) { throw e; } } } if (!renamed) { throw new IOException(String.format("Failed to rename %s to %s.", fromFile.getPath(), toFilePath)); } log.info(String.format("Renamed %s to %s", fromFile.getPath(), toFilePath)); } else { log.info(String.format("File already exists %s. Will not rewrite", toFilePath)); } } }
HadoopUtils { public static void safeRenameRecursively(FileSystem fileSystem, Path from, Path to) throws IOException { for (FileStatus fromFile : FileListUtils.listFilesRecursively(fileSystem, from)) { Path relativeFilePath = new Path(StringUtils.substringAfter(fromFile.getPath().toString(), from.toString() + Path.SEPARATOR)); Path toFilePath = new Path(to, relativeFilePath); if (!fileSystem.exists(toFilePath)) { boolean renamed = false; for (int i = 0; !renamed && i < MAX_RENAME_TRIES; i++) { try { renamed = fileSystem.rename(fromFile.getPath(), toFilePath); break; } catch (FileNotFoundException e) { if (i + 1 >= MAX_RENAME_TRIES) { throw e; } } } if (!renamed) { throw new IOException(String.format("Failed to rename %s to %s.", fromFile.getPath(), toFilePath)); } log.info(String.format("Renamed %s to %s", fromFile.getPath(), toFilePath)); } else { log.info(String.format("File already exists %s. Will not rewrite", toFilePath)); } } } }
HadoopUtils { public static void safeRenameRecursively(FileSystem fileSystem, Path from, Path to) throws IOException { for (FileStatus fromFile : FileListUtils.listFilesRecursively(fileSystem, from)) { Path relativeFilePath = new Path(StringUtils.substringAfter(fromFile.getPath().toString(), from.toString() + Path.SEPARATOR)); Path toFilePath = new Path(to, relativeFilePath); if (!fileSystem.exists(toFilePath)) { boolean renamed = false; for (int i = 0; !renamed && i < MAX_RENAME_TRIES; i++) { try { renamed = fileSystem.rename(fromFile.getPath(), toFilePath); break; } catch (FileNotFoundException e) { if (i + 1 >= MAX_RENAME_TRIES) { throw e; } } } if (!renamed) { throw new IOException(String.format("Failed to rename %s to %s.", fromFile.getPath(), toFilePath)); } log.info(String.format("Renamed %s to %s", fromFile.getPath(), toFilePath)); } else { log.info(String.format("File already exists %s. Will not rewrite", toFilePath)); } } }  }
HadoopUtils { public static void safeRenameRecursively(FileSystem fileSystem, Path from, Path to) throws IOException { for (FileStatus fromFile : FileListUtils.listFilesRecursively(fileSystem, from)) { Path relativeFilePath = new Path(StringUtils.substringAfter(fromFile.getPath().toString(), from.toString() + Path.SEPARATOR)); Path toFilePath = new Path(to, relativeFilePath); if (!fileSystem.exists(toFilePath)) { boolean renamed = false; for (int i = 0; !renamed && i < MAX_RENAME_TRIES; i++) { try { renamed = fileSystem.rename(fromFile.getPath(), toFilePath); break; } catch (FileNotFoundException e) { if (i + 1 >= MAX_RENAME_TRIES) { throw e; } } } if (!renamed) { throw new IOException(String.format("Failed to rename %s to %s.", fromFile.getPath(), toFilePath)); } log.info(String.format("Renamed %s to %s", fromFile.getPath(), toFilePath)); } else { log.info(String.format("File already exists %s. Will not rewrite", toFilePath)); } } }  static Configuration newConfiguration(); @Deprecated static List<FileStatus> listStatusRecursive(FileSystem fileSystem, Path path); static String toUriPath(Path path); static void deletePath(FileSystem fs, Path f, boolean recursive); static void deleteDirectories(FileSystem fs, List<String> directoriesToDelete, boolean recursive, boolean moveToTrash); static void deleteIfExists(FileSystem fs, Path path, boolean recursive); static void deletePathAndEmptyAncestors(FileSystem fs, Path f, boolean recursive); static void deletePathByRegex(FileSystem fs, final Path path, final String regex); static void moveToTrash(FileSystem fs, Path path); static boolean renamePathHandleLocalFSRace(FileSystem fs, Path src, Path dst); static void renamePath(FileContext fc, Path oldName, Path newName); static void renamePath(FileContext fc, Path oldName, Path newName, boolean overwrite); static void renamePath(FileSystem fs, Path oldName, Path newName); static void renamePath(FileSystem fs, Path oldName, Path newName, boolean overwrite); static void movePath(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, Configuration conf); static void movePath(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, boolean overwrite,
      Configuration conf); static void copyPath(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, Configuration conf); static void copyPath(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, boolean overwrite,
      Configuration conf); static void copyFile(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, Path tmp, boolean overwriteDst,
      Configuration conf); static void copyFile(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, boolean overwrite,
      Configuration conf); static void renameRecursively(FileSystem fileSystem, Path from, Path to); static FileSystem getOptionallyThrottledFileSystem(FileSystem fs, State state); static FileSystem getOptionallyThrottledFileSystem(FileSystem fs, int qpsLimit); synchronized static boolean safeRenameIfNotExists(FileSystem fs, Path from, Path to); static boolean unsafeRenameIfNotExists(FileSystem fs, Path from, Path to); static void safeRenameRecursively(FileSystem fileSystem, Path from, Path to); static Configuration getConfFromState(State state); static Configuration getConfFromState(State state, Optional<String> encryptedPath); static Configuration getConfFromProperties(Properties properties); static State getStateFromConf(Configuration conf); static void setGroup(FileSystem fs, Path path, String group); static String serializeToString(Writable writable); static Writable deserializeFromString(Class<? extends Writable> writableClass, String serializedWritableStr); static Writable deserializeFromString(Class<? extends Writable> writableClass, String serializedWritableStr,
      Configuration configuration); static void serializeWriterFilePermissions(State state, int numBranches, int branchId,
      FsPermission fsPermissions); static void serializeWriterDirPermissions(State state, int numBranches, int branchId,
      FsPermission fsPermissions); static void setWriterFileOctalPermissions(State state, int numBranches, int branchId,
      String octalPermissions); static void setWriterDirOctalPermissions(State state, int numBranches, int branchId, String octalPermissions); static FsPermission deserializeWriterFilePermissions(State state, int numBranches, int branchId); static FsPermission deserializeWriterDirPermissions(State state, int numBranches, int branchId); static FsPermission deserializeFsPermission(State props, String propName, FsPermission defaultPermission); static String sanitizePath(String path, String substitute); static Path sanitizePath(Path path, String substitute); static void setPermissions(Path location, Optional<String> owner, Optional<String> group, FileSystem fs,
      FsPermission permission); static boolean hasContent(FileSystem fs, Path path); static void addGobblinSite(); static FileSystem getSourceFileSystem(State state); static FileSystem getWriterFileSystem(State state, int numBranches, int branchId); }
HadoopUtils { public static void safeRenameRecursively(FileSystem fileSystem, Path from, Path to) throws IOException { for (FileStatus fromFile : FileListUtils.listFilesRecursively(fileSystem, from)) { Path relativeFilePath = new Path(StringUtils.substringAfter(fromFile.getPath().toString(), from.toString() + Path.SEPARATOR)); Path toFilePath = new Path(to, relativeFilePath); if (!fileSystem.exists(toFilePath)) { boolean renamed = false; for (int i = 0; !renamed && i < MAX_RENAME_TRIES; i++) { try { renamed = fileSystem.rename(fromFile.getPath(), toFilePath); break; } catch (FileNotFoundException e) { if (i + 1 >= MAX_RENAME_TRIES) { throw e; } } } if (!renamed) { throw new IOException(String.format("Failed to rename %s to %s.", fromFile.getPath(), toFilePath)); } log.info(String.format("Renamed %s to %s", fromFile.getPath(), toFilePath)); } else { log.info(String.format("File already exists %s. Will not rewrite", toFilePath)); } } }  static Configuration newConfiguration(); @Deprecated static List<FileStatus> listStatusRecursive(FileSystem fileSystem, Path path); static String toUriPath(Path path); static void deletePath(FileSystem fs, Path f, boolean recursive); static void deleteDirectories(FileSystem fs, List<String> directoriesToDelete, boolean recursive, boolean moveToTrash); static void deleteIfExists(FileSystem fs, Path path, boolean recursive); static void deletePathAndEmptyAncestors(FileSystem fs, Path f, boolean recursive); static void deletePathByRegex(FileSystem fs, final Path path, final String regex); static void moveToTrash(FileSystem fs, Path path); static boolean renamePathHandleLocalFSRace(FileSystem fs, Path src, Path dst); static void renamePath(FileContext fc, Path oldName, Path newName); static void renamePath(FileContext fc, Path oldName, Path newName, boolean overwrite); static void renamePath(FileSystem fs, Path oldName, Path newName); static void renamePath(FileSystem fs, Path oldName, Path newName, boolean overwrite); static void movePath(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, Configuration conf); static void movePath(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, boolean overwrite,
      Configuration conf); static void copyPath(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, Configuration conf); static void copyPath(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, boolean overwrite,
      Configuration conf); static void copyFile(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, Path tmp, boolean overwriteDst,
      Configuration conf); static void copyFile(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, boolean overwrite,
      Configuration conf); static void renameRecursively(FileSystem fileSystem, Path from, Path to); static FileSystem getOptionallyThrottledFileSystem(FileSystem fs, State state); static FileSystem getOptionallyThrottledFileSystem(FileSystem fs, int qpsLimit); synchronized static boolean safeRenameIfNotExists(FileSystem fs, Path from, Path to); static boolean unsafeRenameIfNotExists(FileSystem fs, Path from, Path to); static void safeRenameRecursively(FileSystem fileSystem, Path from, Path to); static Configuration getConfFromState(State state); static Configuration getConfFromState(State state, Optional<String> encryptedPath); static Configuration getConfFromProperties(Properties properties); static State getStateFromConf(Configuration conf); static void setGroup(FileSystem fs, Path path, String group); static String serializeToString(Writable writable); static Writable deserializeFromString(Class<? extends Writable> writableClass, String serializedWritableStr); static Writable deserializeFromString(Class<? extends Writable> writableClass, String serializedWritableStr,
      Configuration configuration); static void serializeWriterFilePermissions(State state, int numBranches, int branchId,
      FsPermission fsPermissions); static void serializeWriterDirPermissions(State state, int numBranches, int branchId,
      FsPermission fsPermissions); static void setWriterFileOctalPermissions(State state, int numBranches, int branchId,
      String octalPermissions); static void setWriterDirOctalPermissions(State state, int numBranches, int branchId, String octalPermissions); static FsPermission deserializeWriterFilePermissions(State state, int numBranches, int branchId); static FsPermission deserializeWriterDirPermissions(State state, int numBranches, int branchId); static FsPermission deserializeFsPermission(State props, String propName, FsPermission defaultPermission); static String sanitizePath(String path, String substitute); static Path sanitizePath(Path path, String substitute); static void setPermissions(Path location, Optional<String> owner, Optional<String> group, FileSystem fs,
      FsPermission permission); static boolean hasContent(FileSystem fs, Path path); static void addGobblinSite(); static FileSystem getSourceFileSystem(State state); static FileSystem getWriterFileSystem(State state, int numBranches, int branchId); static final String HDFS_ILLEGAL_TOKEN_REGEX; static final Collection<String> FS_SCHEMES_NON_ATOMIC; static final String MAX_FILESYSTEM_QPS; }