@VisibleForTesting
  protected Boolean findNextCompactionAndExecute(boolean collectGenericStats, boolean collectMrStats) {
    // Make sure nothing escapes this run method and kills the metastore at large,
    // so wrap it in a big catch Throwable statement.
    PerfLogger perfLogger = SessionState.getPerfLogger(false);
    String workerMetric = null;
    CompactionInfo ci = null;
    Table table = null;
    CompactionService compactionService = null;
    boolean success = false;

    // If an exception is thrown in the try-with-resources block below, msc is closed and nulled, so a new instance
    // is need to be obtained here.
    if (msc == null) {
      try {
        msc = HiveMetaStoreUtils.getHiveMetastoreClient(conf);
      } catch (Exception e) {
        LOG.error("Failed to connect to HMS", e);
        return false;
      }
    }

    try {

      FindNextCompactRequest findNextCompactRequest = new FindNextCompactRequest();
      findNextCompactRequest.setWorkerId(workerName);
      findNextCompactRequest.setWorkerVersion(runtimeVersion);
      findNextCompactRequest.setPoolName(this.getPoolName());
      ci = CompactionInfo.optionalCompactionInfoStructToInfo(msc.findNextCompact(findNextCompactRequest));
      LOG.info("Processing compaction request {}", ci);

      if (ci == null) {
        return false;
      }
      if ((runtimeVersion == null && ci.initiatorVersion != null) || (runtimeVersion != null && !runtimeVersion.equals(ci.initiatorVersion))) {
        LOG.warn("Worker and Initiator versions do not match. Worker: v{}, Initiator: v{}", runtimeVersion, ci.initiatorVersion);
      }

      if (StringUtils.isBlank(getPoolName()) && StringUtils.isNotBlank(ci.poolName)) {
        LOG.warn("A timed out copmaction pool entry ({}) is picked up by one of the default compaction pool workers.", ci);
      }
      if (StringUtils.isNotBlank(getPoolName()) && StringUtils.isNotBlank(ci.poolName) && !getPoolName().equals(ci.poolName)) {
        LOG.warn("The returned compaction request ({}) belong to a different pool. Although the worker is assigned to the {} pool," +
            " it will process the request.", ci, getPoolName());
      }
      CompactorUtil.checkInterrupt(CLASS_NAME);

      if (MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.METASTORE_ACIDMETRICS_EXT_ON)) {
        workerMetric = MetricsConstants.COMPACTION_WORKER_CYCLE + "_" +
            (ci.type != null ? ci.type.toString().toLowerCase() : null);
        perfLogger.perfLogBegin(CLASS_NAME, workerMetric);
      }

      // Find the table we will be working with.
      try {
        table = resolveTable(ci);
        if (table == null) {
          ci.errorMessage = "Unable to find table " + ci.getFullTableName() + ", assuming it was dropped and moving on.";
          LOG.warn(ci.errorMessage + " Compaction info: {}", ci);
          msc.markRefused(CompactionInfo.compactionInfoToStruct(ci));
          return false;
        }
      } catch (MetaException e) {
        LOG.error("Unexpected error during resolving table. Compaction info: " + ci, e);
        ci.errorMessage = e.getMessage();
        msc.markFailed(CompactionInfo.compactionInfoToStruct(ci));
        return false;
      }

      CompactorUtil.checkInterrupt(CLASS_NAME);
      compactionService = CompactionExecutorFactory.createExecutor(conf, msc, compactorFactory, table, collectGenericStats, collectMrStats);

      try {
        success = compactionService.compact(table, ci);
      } catch (Throwable e) {
        LOG.error("Caught exception while trying to compact " + ci +
            ". Marking failed to avoid repeated failures", e);
        markFailed(ci, e.getMessage());

        if (CompactorUtil.runJobAsSelf(ci.runAs)) {
          compactionService.cleanupResultDirs(ci);
        } else {
          LOG.info("Cleaning as user " + ci.runAs);
          UserGroupInformation ugi = UserGroupInformation.createProxyUser(ci.runAs,
              UserGroupInformation.getLoginUser());

          CompactionService finalCompactionService = compactionService;
          CompactionInfo finalCi = ci;
          ugi.doAs((PrivilegedExceptionAction<Void>) () -> {
            finalCompactionService.cleanupResultDirs(finalCi);
            return null;
          });
          try {
            FileSystem.closeAllForUGI(ugi);
          } catch (IOException ex) {
            LOG.error("Could not clean up file-system handles for UGI: " + ugi, e);
          }
        }
      }
    } catch (TException | IOException t) {
      LOG.error("Caught an exception in the main loop of compactor worker " + workerName, t);

      markFailed(ci, t.getMessage());

      if (msc != null) {
        msc.close();
        msc = null;
      }
    } catch (Throwable t) {
      LOG.error("Caught an exception in the main loop of compactor worker " + workerName, t);
    } finally {
      if (workerMetric != null && MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.METASTORE_ACIDMETRICS_EXT_ON)) {
        perfLogger.perfLogEnd(CLASS_NAME, workerMetric);
      }
    }

    if (success && compactionService.computeStats()) {
      statsUpdater.gatherStats(
          conf,
          ci,
          table.getParameters(),
          CompactorUtil.runJobAsSelf(ci.runAs) ? ci.runAs : table.getOwner(),
          CompactorUtil.getCompactorJobQueueName(conf, ci, table),
          msc);
    }
    return success;
  }
Finds the next compaction and executes it. The main thread might interrupt the execution of this method
in case of timeout.
@param collectGenericStats If true then for both MR and Query based compaction the stats are regenerated
@param collectMrStats If true then for MR compaction the stats are regenerated
@return Returns true, if there was compaction in the queue, and we started working on it.
ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java