@Override
  public void run() {
    LOG.info("Starting Initiator thread");
    // Make sure nothing escapes this run method and kills the metastore at large,
    // so wrap it in a big catch Throwable statement.
    try {
      recoverFailedCompactions(false);

      int abortedThreshold = HiveConf.getIntVar(conf,
          HiveConf.ConfVars.HIVE_COMPACTOR_ABORTEDTXN_THRESHOLD);
      long abortedTimeThreshold = HiveConf
          .getTimeVar(conf, HiveConf.ConfVars.HIVE_COMPACTOR_ABORTEDTXN_TIME_THRESHOLD,
              TimeUnit.MILLISECONDS);
      TxnStore.MutexAPI mutex = shouldUseMutex ? txnHandler.getMutexAPI() : new NoMutex();

      // Make sure we run through the loop once before checking to stop as this makes testing
      // much easier.  The stop value is only for testing anyway and not used when called from
      // HiveMetaStore.
      do {
        PerfLogger perfLogger = PerfLogger.getPerfLogger(false);
        long startedAt = -1;
        long prevStart;

        // Wrap the inner parts of the loop in a catch throwable so that any errors in the loop
        // don't doom the entire thread.
        try (TxnStore.MutexAPI.LockHandle handle = mutex.acquireLock(TxnStore.MUTEX_KEY.Initiator.name())) {
          startedAt = System.currentTimeMillis();
          prevStart = handle.getLastUpdateTime();

          if (metricsEnabled) {
            perfLogger.perfLogBegin(CLASS_NAME, MetricsConstants.COMPACTION_INITIATOR_CYCLE);
            stopCycleUpdater();
            startCycleUpdater(HiveConf.getTimeVar(conf,
                    HiveConf.ConfVars.HIVE_COMPACTOR_INITIATOR_DURATION_UPDATE_INTERVAL, TimeUnit.MILLISECONDS),
                new InitiatorCycleUpdater(MetricsConstants.COMPACTION_INITIATOR_CYCLE_DURATION,
                    startedAt,
                    MetastoreConf.getTimeVar(conf,
                        MetastoreConf.ConfVars.COMPACTOR_LONG_RUNNING_INITIATOR_THRESHOLD_WARNING,
                        TimeUnit.MILLISECONDS),
                    MetastoreConf.getTimeVar(conf,
                        MetastoreConf.ConfVars.COMPACTOR_LONG_RUNNING_INITIATOR_THRESHOLD_ERROR,
                        TimeUnit.MILLISECONDS)));
          }

          final ShowCompactResponse currentCompactions = txnHandler.showCompact(new ShowCompactRequest());

          CompactorUtil.checkInterrupt(CLASS_NAME);

          // Currently we invalidate all entries after each cycle, because the bootstrap replication is marked via
          // table property hive.repl.first.inc.pending which would be cached.
          metadataCache.invalidate();
          Set<String> skipDBs = Sets.newConcurrentHashSet();
          Set<String> skipTables = Sets.newConcurrentHashSet();

          Set<CompactionInfo> potentials = compactionExecutor.submit(() ->
            txnHandler.findPotentialCompactions(abortedThreshold, abortedTimeThreshold, prevStart)
              .parallelStream()
              .filter(ci -> isEligibleForCompaction(ci, currentCompactions, skipDBs, skipTables))
              .collect(Collectors.toSet())).get();
          LOG.debug("Found {} potential compactions, checking to see if we should compact any of them", potentials.size());

          CompactorUtil.checkInterrupt(CLASS_NAME);

          Map<String, String> tblNameOwners = new HashMap<>();
          List<CompletableFuture<Void>> compactionList = new ArrayList<>();

          if (!potentials.isEmpty()) {
            ValidTxnList validTxnList = TxnCommonUtils.createValidReadTxnList(
                txnHandler.getOpenTxns(), 0);
            conf.set(ValidTxnList.VALID_TXNS_KEY, validTxnList.writeToString());
          }

          for (CompactionInfo ci : potentials) {
            try {
              //Check for interruption before scheduling each compactionInfo and return if necessary
              if (Thread.currentThread().isInterrupted()) {
                return;
              }

              Table t = metadataCache.computeIfAbsent(ci.getFullTableName(), () -> resolveTable(ci));
              ci.poolName = CompactorUtil.getPoolName(conf, t, metadataCache);
              Partition p = resolvePartition(ci);
              if (p == null && ci.partName != null) {
                LOG.info("Can't find partition " + ci.getFullPartitionName() +
                    ", assuming it has been dropped and moving on.");
                continue;
              }
              String runAs = resolveUserToRunAs(tblNameOwners, t, p);
              /* checkForCompaction includes many file metadata checks and may be expensive.
               * Therefore, using a thread pool here and running checkForCompactions in parallel */
              String tableName = ci.getFullTableName();
              String partition = ci.getFullPartitionName();
              ci.initiatorVersion = this.runtimeVersion;
              CompletableFuture<Void> asyncJob =
                  CompletableFuture.runAsync(
                          CompactorUtil.ThrowingRunnable.unchecked(() ->
                                  CompactorUtil.scheduleCompactionIfRequired(ci, t, p, runAs, metricsEnabled, hostName, txnHandler, conf)), compactionExecutor)
                      .exceptionally(exc -> {
                        LOG.error("Error while running scheduling the compaction on the table {} / partition {}.", tableName, partition, exc);
                        return null;
                      });
              compactionList.add(asyncJob);
            } catch (Throwable t) {
              LOG.error("Caught exception while trying to determine if we should compact {}. " +
                  "Marking failed to avoid repeated failures, {}", ci, t);
              ci.errorMessage = t.getMessage();
              txnHandler.markFailed(ci);
            }
          }

          //Use get instead of join, so we can receive InterruptedException and shutdown gracefully
          CompletableFuture.allOf(compactionList.toArray(new CompletableFuture[0])).get();

          // Check for timed out remote workers.
          recoverFailedCompactions(true);
          handle.releaseLocks(startedAt);
        } catch (InterruptedException e) {
          // do not ignore interruption requests
          return;
        } catch (Throwable t) {
          LOG.error("Initiator loop caught unexpected exception this time through the loop", t);
        } finally {
          if (metricsEnabled) {
            perfLogger.perfLogEnd(CLASS_NAME, MetricsConstants.COMPACTION_INITIATOR_CYCLE);
            updateCycleDurationMetric(MetricsConstants.COMPACTION_INITIATOR_CYCLE_DURATION, startedAt);
          }
          stopCycleUpdater();
        }

        doPostLoopActions(System.currentTimeMillis() - startedAt);
      } while (!stop.get());
    } catch (Throwable t) {
      LOG.error("Caught an exception in the main loop of compactor initiator, exiting.", t);
    } finally {
      if (Thread.currentThread().isInterrupted()) {
        LOG.info("Interrupt received, Initiator is shutting down.");
      }
      if (compactionExecutor != null) {
        compactionExecutor.shutdownNow();
      }
    }
  }
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