private void clean(CompactionInfo ci, long minOpenTxn, boolean metricsEnabled) throws MetaException {
    LOG.info("Starting cleaning for {}", ci);
    PerfLogger perfLogger = PerfLogger.getPerfLogger(false);
    String cleanerMetric = MetricsConstants.COMPACTION_CLEANER_CYCLE + "_" +
            (!isNull(ci.type) ? ci.type.toString().toLowerCase() : null);
    try {
      if (metricsEnabled) {
        perfLogger.perfLogBegin(CompactionCleaner.class.getName(), cleanerMetric);
      }
      final String location = ci.getProperty("location");

      Table t = null;
      Partition p = null;

      if (isNull(location)) {
        t = metadataCache.computeIfAbsent(ci.getFullTableName(), () -> resolveTable(ci.dbname, ci.tableName));
        if (isNull(t)) {
          // The table was dropped before we got around to cleaning it.
          LOG.info("Unable to find table {}, assuming it was dropped. {}", ci.getFullTableName(),
                  idWatermark(ci));
          txnHandler.markCleaned(ci);
          return;
        }
        if (MetaStoreUtils.isNoCleanUpSet(t.getParameters())) {
          // The table was marked no clean up true.
          LOG.info("Skipping table {} clean up, as NO_CLEANUP set to true", ci.getFullTableName());
          txnHandler.markRefused(ci);
          return;
        }
        if (!isNull(ci.partName)) {
          p = resolvePartition(ci.dbname, ci.tableName, ci.partName);
          if (isNull(p)) {
            // The partition was dropped before we got around to cleaning it.
            LOG.info("Unable to find partition {}, assuming it was dropped. {}",
                    ci.getFullPartitionName(), idWatermark(ci));
            txnHandler.markCleaned(ci);
            return;
          }
          if (MetaStoreUtils.isNoCleanUpSet(p.getParameters())) {
            // The partition was marked no clean up true.
            LOG.info("Skipping partition {} clean up, as NO_CLEANUP set to true", ci.getFullPartitionName());
            txnHandler.markRefused(ci);
            return;
          }
        }
      }
      txnHandler.markCleanerStart(ci);

      if (!isNull(t) || !isNull(ci.partName)) {
        String path = isNull(location)
                ? CompactorUtil.resolveStorageDescriptor(t, p).getLocation()
                : location;
        boolean dropPartition = !isNull(ci.partName) && isNull(p);

        //check if partition wasn't re-created
        if (dropPartition && isNull(resolvePartition(ci.dbname, ci.tableName, ci.partName))) {
          cleanUsingLocation(ci, path, true);
        } else {
          cleanUsingAcidDir(ci, path, minOpenTxn);
        }
      } else {
        cleanUsingLocation(ci, location, false);
      }
    } catch (Exception e) {
      LOG.error("Caught exception when cleaning, unable to complete cleaning of {} due to {}", ci,
              e.getMessage());
      if (metricsEnabled) {
        Metrics.getOrCreateCounter(MetricsConstants.COMPACTION_CLEANER_FAILURE_COUNTER).inc();
      }
      handleCleanerAttemptFailure(ci, e.getMessage());
    }  finally {
      if (metricsEnabled) {
        perfLogger.perfLogEnd(CompactionCleaner.class.getName(), cleanerMetric);
      }
    }
  }
A compaction based implementation of TaskHandler.
Provides implementation of creation of compaction clean tasks.
ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/handler/CompactionCleaner.java