private boolean foundCurrentOrFailedCompactions(ShowCompactResponse compactions, CompactionInfo ci) throws MetaException {
    if (compactions.getCompacts() == null) {
      return false;
    }

    //In case of an aborted Dynamic partition insert, the created entry in the compaction queue does not contain
    //a partition name even for partitioned tables. As a result it can happen that the ShowCompactResponse contains
    //an element without partition name for partitioned tables. Therefore, it is necessary to null check the partition
    //name of the ShowCompactResponseElement even if the CompactionInfo.partName is not null. These special compaction
    //requests are skipped by the worker, and only cleaner will pick them up, so we should allow to schedule a 'normal'
    //compaction for partitions of those tables which has special (DP abort) entry with undefined partition name.
    List<ShowCompactResponseElement> filteredElements = compactions.getCompacts().stream()
      .filter(e -> e.getDbname().equals(ci.dbname)
        && e.getTablename().equals(ci.tableName)
        && (e.getPartitionname() == null && ci.partName == null ||
              (Objects.equals(e.getPartitionname(),ci.partName))))
      .collect(Collectors.toList());

    // Figure out if there are any currently running compactions on the same table or partition.
    if (filteredElements.stream().anyMatch(
        e -> TxnStore.WORKING_RESPONSE.equals(e.getState()) || TxnStore.INITIATED_RESPONSE.equals(e.getState()))) {

      LOG.info("Found currently initiated or working compaction for " +
        ci.getFullPartitionName() + " so we will not initiate another compaction");
      return true;
    }

    // Check if there is already sufficient number of consecutive failures for this table/partition
    // so that no new automatic compactions needs to be scheduled.
    int failedThreshold = MetastoreConf.getIntVar(conf, MetastoreConf.ConfVars.COMPACTOR_INITIATOR_FAILED_THRESHOLD);

    LongSummaryStatistics failedStats = filteredElements.stream()
      .filter(e -> TxnStore.SUCCEEDED_RESPONSE.equals(e.getState()) || TxnStore.FAILED_RESPONSE.equals(e.getState()))
      .sorted(Comparator.comparingLong(ShowCompactResponseElement::getId).reversed())
      .limit(failedThreshold)

      .filter(e -> TxnStore.FAILED_RESPONSE.equals(e.getState()))
      .collect(Collectors.summarizingLong(ShowCompactResponseElement::getEnqueueTime));

    // If the last attempt was too long ago, ignore the failed threshold and try compaction again
    long retryTime = MetastoreConf.getTimeVar(conf,
      MetastoreConf.ConfVars.COMPACTOR_INITIATOR_FAILED_RETRY_TIME, TimeUnit.MILLISECONDS);

    boolean needsRetry = (retryTime > 0) && (failedStats.getMax() + retryTime < System.currentTimeMillis());
    if (failedStats.getCount() == failedThreshold && !needsRetry) {
      LOG.warn("Will not initiate compaction for " + ci.getFullPartitionName() + " since last " +
        MetastoreConf.ConfVars.COMPACTOR_INITIATOR_FAILED_THRESHOLD + " attempts to compact it failed.");

      ci.errorMessage = "Compaction is not initiated since last " +
        MetastoreConf.ConfVars.COMPACTOR_INITIATOR_FAILED_THRESHOLD + " consecutive compaction attempts failed)";

      txnHandler.markFailed(ci);
      return true;
    }
    return false;
  }
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