@Test public void testRenameRecursively() throws Exception { final Path hadoopUtilsTestDir = new Path(Files.createTempDir().getAbsolutePath(), "HadoopUtilsTestDir"); FileSystem fs = FileSystem.getLocal(new Configuration()); try { fs.mkdirs(hadoopUtilsTestDir); fs.mkdirs(new Path(hadoopUtilsTestDir, "testRename/a/b/c")); fs.mkdirs(new Path(hadoopUtilsTestDir, "testRenameStaging/a/b/c")); fs.mkdirs(new Path(hadoopUtilsTestDir, "testRenameStaging/a/b/c/e")); fs.create(new Path(hadoopUtilsTestDir, "testRenameStaging/a/b/c/t1.txt")); fs.create(new Path(hadoopUtilsTestDir, "testRenameStaging/a/b/c/e/t2.txt")); HadoopUtils.renameRecursively(fs, new Path(hadoopUtilsTestDir, "testRenameStaging"), new Path(hadoopUtilsTestDir, "testRename")); Assert.assertTrue(fs.exists(new Path(hadoopUtilsTestDir, "testRename/a/b/c/t1.txt"))); Assert.assertTrue(fs.exists(new Path(hadoopUtilsTestDir, "testRename/a/b/c/e/t2.txt"))); } finally { fs.delete(hadoopUtilsTestDir, true); } }
public static void renameRecursively(FileSystem fileSystem, Path from, Path to) throws IOException { log.info(String.format("Recursively renaming %s in %s to %s.", from, fileSystem.getUri(), to)); FileSystem throttledFS = getOptionallyThrottledFileSystem(fileSystem, 10000); ExecutorService executorService = ScalingThreadPoolExecutor.newScalingThreadPool(1, 100, 100, ExecutorsUtils.newThreadFactory(Optional.of(log), Optional.of("rename-thread-%d"))); Queue<Future<?>> futures = Queues.newConcurrentLinkedQueue(); try { if (!fileSystem.exists(from)) { throw new IOException("Trying to rename a path that does not exist! " + from); } futures.add(executorService .submit(new RenameRecursively(throttledFS, fileSystem.getFileStatus(from), to, executorService, futures))); int futuresUsed = 0; while (!futures.isEmpty()) { try { futures.poll().get(); futuresUsed++; } catch (ExecutionException | InterruptedException ee) { throw new IOException(ee.getCause()); } } log.info(String.format("Recursive renaming of %s to %s. (details: used %d futures)", from, to, futuresUsed)); } finally { ExecutorsUtils.shutdownExecutorService(executorService, Optional.of(log), 1, TimeUnit.SECONDS); } }
HadoopUtils { public static void renameRecursively(FileSystem fileSystem, Path from, Path to) throws IOException { log.info(String.format("Recursively renaming %s in %s to %s.", from, fileSystem.getUri(), to)); FileSystem throttledFS = getOptionallyThrottledFileSystem(fileSystem, 10000); ExecutorService executorService = ScalingThreadPoolExecutor.newScalingThreadPool(1, 100, 100, ExecutorsUtils.newThreadFactory(Optional.of(log), Optional.of("rename-thread-%d"))); Queue<Future<?>> futures = Queues.newConcurrentLinkedQueue(); try { if (!fileSystem.exists(from)) { throw new IOException("Trying to rename a path that does not exist! " + from); } futures.add(executorService .submit(new RenameRecursively(throttledFS, fileSystem.getFileStatus(from), to, executorService, futures))); int futuresUsed = 0; while (!futures.isEmpty()) { try { futures.poll().get(); futuresUsed++; } catch (ExecutionException | InterruptedException ee) { throw new IOException(ee.getCause()); } } log.info(String.format("Recursive renaming of %s to %s. (details: used %d futures)", from, to, futuresUsed)); } finally { ExecutorsUtils.shutdownExecutorService(executorService, Optional.of(log), 1, TimeUnit.SECONDS); } } }
HadoopUtils { public static void renameRecursively(FileSystem fileSystem, Path from, Path to) throws IOException { log.info(String.format("Recursively renaming %s in %s to %s.", from, fileSystem.getUri(), to)); FileSystem throttledFS = getOptionallyThrottledFileSystem(fileSystem, 10000); ExecutorService executorService = ScalingThreadPoolExecutor.newScalingThreadPool(1, 100, 100, ExecutorsUtils.newThreadFactory(Optional.of(log), Optional.of("rename-thread-%d"))); Queue<Future<?>> futures = Queues.newConcurrentLinkedQueue(); try { if (!fileSystem.exists(from)) { throw new IOException("Trying to rename a path that does not exist! " + from); } futures.add(executorService .submit(new RenameRecursively(throttledFS, fileSystem.getFileStatus(from), to, executorService, futures))); int futuresUsed = 0; while (!futures.isEmpty()) { try { futures.poll().get(); futuresUsed++; } catch (ExecutionException | InterruptedException ee) { throw new IOException(ee.getCause()); } } log.info(String.format("Recursive renaming of %s to %s. (details: used %d futures)", from, to, futuresUsed)); } finally { ExecutorsUtils.shutdownExecutorService(executorService, Optional.of(log), 1, TimeUnit.SECONDS); } }  }
HadoopUtils { public static void renameRecursively(FileSystem fileSystem, Path from, Path to) throws IOException { log.info(String.format("Recursively renaming %s in %s to %s.", from, fileSystem.getUri(), to)); FileSystem throttledFS = getOptionallyThrottledFileSystem(fileSystem, 10000); ExecutorService executorService = ScalingThreadPoolExecutor.newScalingThreadPool(1, 100, 100, ExecutorsUtils.newThreadFactory(Optional.of(log), Optional.of("rename-thread-%d"))); Queue<Future<?>> futures = Queues.newConcurrentLinkedQueue(); try { if (!fileSystem.exists(from)) { throw new IOException("Trying to rename a path that does not exist! " + from); } futures.add(executorService .submit(new RenameRecursively(throttledFS, fileSystem.getFileStatus(from), to, executorService, futures))); int futuresUsed = 0; while (!futures.isEmpty()) { try { futures.poll().get(); futuresUsed++; } catch (ExecutionException | InterruptedException ee) { throw new IOException(ee.getCause()); } } log.info(String.format("Recursive renaming of %s to %s. (details: used %d futures)", from, to, futuresUsed)); } finally { ExecutorsUtils.shutdownExecutorService(executorService, Optional.of(log), 1, TimeUnit.SECONDS); } }  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 renameRecursively(FileSystem fileSystem, Path from, Path to) throws IOException { log.info(String.format("Recursively renaming %s in %s to %s.", from, fileSystem.getUri(), to)); FileSystem throttledFS = getOptionallyThrottledFileSystem(fileSystem, 10000); ExecutorService executorService = ScalingThreadPoolExecutor.newScalingThreadPool(1, 100, 100, ExecutorsUtils.newThreadFactory(Optional.of(log), Optional.of("rename-thread-%d"))); Queue<Future<?>> futures = Queues.newConcurrentLinkedQueue(); try { if (!fileSystem.exists(from)) { throw new IOException("Trying to rename a path that does not exist! " + from); } futures.add(executorService .submit(new RenameRecursively(throttledFS, fileSystem.getFileStatus(from), to, executorService, futures))); int futuresUsed = 0; while (!futures.isEmpty()) { try { futures.poll().get(); futuresUsed++; } catch (ExecutionException | InterruptedException ee) { throw new IOException(ee.getCause()); } } log.info(String.format("Recursive renaming of %s to %s. (details: used %d futures)", from, to, futuresUsed)); } finally { ExecutorsUtils.shutdownExecutorService(executorService, Optional.of(log), 1, TimeUnit.SECONDS); } }  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; }