private List<Path> removeFiles(CleanupRequest cr)
          throws MetaException, IOException {
    List<Path> deleted = new ArrayList<>();
    if (cr.getObsoleteDirs().isEmpty()) {
      return deleted;
    }
    LOG.info("About to remove {} obsolete directories from {}. {}", cr.getObsoleteDirs().size(),
            cr.getLocation(), CompactorUtil.getDebugInfo(cr.getObsoleteDirs()));
    boolean needCmRecycle;
    try {
      Database db = metadataCache.computeIfAbsent(cr.getDbName(),
              () -> CompactorUtil.resolveDatabase(conf, cr.getDbName()));
      needCmRecycle = ReplChangeManager.isSourceOfReplication(db);
    } catch (NoSuchObjectException ex) {
      // can not drop a database which is a source of replication
      needCmRecycle = false;
    } catch (RuntimeException ex) {
      if (ex.getCause() instanceof NoSuchObjectException) {
        // can not drop a database which is a source of replication
        needCmRecycle = false;
      } else {
        throw ex;
      }
    } catch (Exception ex) {
      throw new MetaException(ex.getMessage());
    }
    FileSystem fs = new Path(cr.getLocation()).getFileSystem(conf);
    for (Path dead : cr.getObsoleteDirs()) {
      LOG.debug("Going to delete path: {}", dead);
      if (needCmRecycle) {
        replChangeManager.recycle(dead, ReplChangeManager.RecycleType.MOVE, cr.isPurge());
      }
      if (FileUtils.moveToTrash(fs, dead, conf, cr.isPurge())) {
        deleted.add(dead);
      }
    }
    return deleted;
  }
@param cr Cleaning request
@return List of deleted files if any files were removed
ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/FSRemover.java