@Test(groups = { "performance" }) public void testRenamePerformance() throws Exception { FileSystem fs = Mockito.mock(FileSystem.class); Path sourcePath = new Path("/source"); Path s1 = new Path(sourcePath, "d1"); FileStatus[] sourceStatuses = new FileStatus[10000]; FileStatus[] targetStatuses = new FileStatus[1000]; for (int i = 0; i < sourceStatuses.length; i++) { sourceStatuses[i] = getFileStatus(new Path(s1, "path" + i), false); } for (int i = 0; i < targetStatuses.length; i++) { targetStatuses[i] = getFileStatus(new Path(s1, "path" + i), false); } Mockito.when(fs.getUri()).thenReturn(new URI("file: Mockito.when(fs.getFileStatus(sourcePath)).thenAnswer(getDelayedAnswer(getFileStatus(sourcePath, true))); Mockito.when(fs.exists(sourcePath)).thenAnswer(getDelayedAnswer(true)); Mockito.when(fs.listStatus(sourcePath)).thenAnswer(getDelayedAnswer(new FileStatus[]{getFileStatus(s1, true)})); Mockito.when(fs.exists(s1)).thenAnswer(getDelayedAnswer(true)); Mockito.when(fs.listStatus(s1)).thenAnswer(getDelayedAnswer(sourceStatuses)); Path target = new Path("/target"); Path s1Target = new Path(target, "d1"); Mockito.when(fs.exists(target)).thenAnswer(getDelayedAnswer(true)); Mockito.when(fs.exists(s1Target)).thenAnswer(getDelayedAnswer(true)); Mockito.when(fs.mkdirs(Mockito.any(Path.class))).thenAnswer(getDelayedAnswer(true)); Mockito.when(fs.rename(Mockito.any(Path.class), Mockito.any(Path.class))).thenAnswer(getDelayedAnswer(true)); HadoopUtils.renameRecursively(fs, sourcePath, target); }
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; }