private boolean isEligibleForCompaction(CompactionInfo ci,
      ShowCompactResponse currentCompactions, Set<String> skipDBs, Set<String> skipTables) {
    try {
      if (skipDBs.contains(ci.dbname)) {
        LOG.info("Skipping {}::{}, skipDBs::size:{}", ci.dbname, ci.tableName, skipDBs.size());
        return false;
      } else {
        if (replIsCompactionDisabledForDatabase(ci.dbname)) {
          skipDBs.add(ci.dbname);
          LOG.info("Skipping {} as compaction is disabled due to repl; skipDBs::size:{}",
              ci.dbname, skipDBs.size());
          return false;
        }
      }

      if (skipTables.contains(ci.getFullTableName())) {
        return false;
      }

      LOG.info("Checking to see if we should compact " + ci.getFullPartitionName());

      // Check if we have already initiated or are working on a compaction for this table/partition.
      // Also make sure we haven't exceeded configured number of consecutive failures.
      // If any of the above applies, skip it.
      // Note: if we are just waiting on cleaning we can still check, as it may be time to compact again even though we haven't cleaned.
      if (foundCurrentOrFailedCompactions(currentCompactions, ci)) {
        return false;
      }

      Table t = metadataCache.computeIfAbsent(ci.getFullTableName(), () -> resolveTable(ci));
      if (t == null) {
        LOG.info("Can't find table " + ci.getFullTableName() + ", assuming it's a temp " +
            "table or has been dropped and moving on.");
        return false;
      }

      if (replIsCompactionDisabledForTable(t)) {
        skipTables.add(ci.getFullTableName());
        return false;
      }

      Map<String, String> dbParams = metadataCache.computeIfAbsent(ci.dbname, () -> resolveDatabase(ci)).getParameters();
      if (MetaStoreUtils.isNoAutoCompactSet(dbParams, t.getParameters())) {
        if (Boolean.parseBoolean(MetaStoreUtils.getNoAutoCompact(dbParams))) {
          skipDBs.add(ci.dbname);
          LOG.info("DB " + ci.dbname + " marked " + hive_metastoreConstants.NO_AUTO_COMPACT +
              "=true so we will not compact it.");
        } else {
          skipTables.add(ci.getFullTableName());
          LOG.info("Table " + tableName(t) + " marked " + hive_metastoreConstants.NO_AUTO_COMPACT +
              "=true so we will not compact it.");
        }
        return false;
      }
      if (AcidUtils.isInsertOnlyTable(t.getParameters()) && !HiveConf
          .getBoolVar(conf, HiveConf.ConfVars.HIVE_COMPACTOR_COMPACT_MM)) {
        skipTables.add(ci.getFullTableName());
        LOG.info("Table " + tableName(t) + " is insert only and " + HiveConf.ConfVars.HIVE_COMPACTOR_COMPACT_MM.varname
            + "=false so we will not compact it.");
        return false;
      }
      if (isDynPartIngest(t, ci)) {
        return false;
      }

    } catch (Throwable e) {
      LOG.error("Caught exception while checking compaction eligibility.", e);
      try {
        ci.errorMessage = e.getMessage();
        txnHandler.markFailed(ci);
      } catch (MetaException ex) {
        LOG.error("Caught exception while marking compaction as failed.", e);
      }
      return false;
    }
    return true;
  }
A class to initiate compactions.  This will run in a separate thread.
It's critical that there exactly 1 of these in a given warehouse.
ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java