private void traverseAndDelete(Path dir, boolean root, CompletableFuture<Boolean> result) {
    try {
      // Step.1: List all files under the given directory.
      List<FileStatus> allPaths = Arrays.asList(fs.listStatus(dir));
      List<FileStatus> subDirs =
        allPaths.stream().filter(FileStatus::isDirectory).collect(Collectors.toList());
      List<FileStatus> files =
        allPaths.stream().filter(FileStatus::isFile).collect(Collectors.toList());

      // Step.2: Try to delete all the deletable files.
      boolean allFilesDeleted =
        files.isEmpty() || deleteAction(() -> checkAndDeleteFiles(files), "files", dir);

      // Step.3: Start to traverse and delete the sub-directories.
      List<CompletableFuture<Boolean>> futures = new ArrayList<>();
      if (!subDirs.isEmpty()) {
        if (sortDirectories) {
          sortByConsumedSpace(subDirs);
        }
        // Submit the request of sub-directory deletion.
        subDirs.forEach(subDir -> {
          if (!shouldExclude(subDir)) {
            CompletableFuture<Boolean> subFuture = new CompletableFuture<>();
            pool.execute(() -> traverseAndDelete(subDir.getPath(), false, subFuture));
            futures.add(subFuture);
          }
        });
      }

      // Step.4: Once all sub-files & sub-directories are deleted, then can try to delete the
      // current directory asynchronously.
      FutureUtils.addListener(
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])),
        (voidObj, e) -> {
          if (e != null) {
            result.completeExceptionally(FutureUtils.unwrapCompletionException(e));
            return;
          }
          try {
            boolean allSubDirsDeleted = futures.stream().allMatch(CompletableFuture::join);
            boolean deleted = allFilesDeleted && allSubDirsDeleted && isEmptyDirDeletable(dir);
            if (deleted && !root) {
              // If and only if files and sub-dirs under current dir are deleted successfully, and
              // the empty directory can be deleted, and it is not the root dir then task will
              // try to delete it.
              deleted = deleteAction(() -> fs.delete(dir, false), "dir", dir);
            }
            result.complete(deleted);
          } catch (Exception ie) {
            // Must handle the inner exception here, otherwise the result may get stuck if one
            // sub-directory get some failure.
            result.completeExceptionally(ie);
          }
        });
    } catch (Exception e) {
      if (e instanceof FileNotFoundException) {
        LOG.debug("Dir dose not exist, {}", dir);
      } else {
        LOG.error("Failed to traverse and delete the path: {}", dir, e);
      }
      result.completeExceptionally(e);
    }
  }
Attempts to clean up a directory(its subdirectories, and files) in a
{@link java.util.concurrent.ThreadPoolExecutor} concurrently. We can get the final result by
calling result.get().
hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/CleanerChore.java