@Override
  public void run() {
    LOG.info("Starting Worker thread");
    boolean genericStats = conf.getBoolVar(HiveConf.ConfVars.HIVE_COMPACTOR_GATHER_STATS);
    boolean mrStats = conf.getBoolVar(HiveConf.ConfVars.HIVE_MR_COMPACTOR_GATHER_STATS);
    long timeout = conf.getTimeVar(HiveConf.ConfVars.HIVE_COMPACTOR_WORKER_TIMEOUT, TimeUnit.MILLISECONDS);
    long nextSleep = SLEEP_TIME;
    boolean launchedJob;
    ExecutorService executor = getTimeoutHandlingExecutor();
    try {
      do {
        long startedAt = System.currentTimeMillis();
        boolean err = false;
        launchedJob = false;
        Future<Boolean> singleRun = executor.submit(() -> findNextCompactionAndExecute(genericStats, mrStats));
        try {
          launchedJob = singleRun.get(timeout, TimeUnit.MILLISECONDS);
        } catch (TimeoutException te) {
          LOG.info("Timeout during executing compaction", te);
          // Cancel the job, and recreate the Executor as well, so we can be sure that we have an available thread
          // even if we can not interrupt the task somehow. (Trade possible resource hogging for compactor stability)
          singleRun.cancel(true);
          executor.shutdownNow();
          executor = getTimeoutHandlingExecutor();
          err = true;
        } catch (ExecutionException e) {
          LOG.info("Exception during executing compaction", e);
          err = true;
        } catch (InterruptedException ie) {
          Thread.currentThread().interrupt();
        } catch (Throwable t) {
          err = true;
        }

        doPostLoopActions(System.currentTimeMillis() - startedAt);

        // If we didn't try to launch a job it either means there was no work to do or we got
        // here as the result of an error like communication failure with the DB, schema failures etc.  Either way we want to wait
        // a bit before, otherwise we can start over the loop immediately.
        if ((!launchedJob || err) && !stop.get()) {
          Thread.sleep(nextSleep);
        }
        //Backoff mechanism
        //Increase sleep time if error persist
        //Reset sleep time to default once error is resolved
        nextSleep = (err) ? nextSleep * 2 : SLEEP_TIME;
        if (nextSleep > SLEEP_TIME_MAX) nextSleep = SLEEP_TIME_MAX;

      } while (!stop.get());
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    } catch (Throwable t) {
      LOG.error("Caught an exception in the main loop of compactor worker, exiting.", t);
    } finally {
      if (Thread.currentThread().isInterrupted()) {
        LOG.info("Interrupt received, Worker is shutting down.");
      }
      executor.shutdownNow();
      if (msc != null) {
        msc.close();
      }
    }
  }
A class to do compactions.  This will run in a separate thread.  It will spin on the
compaction queue and look for new work to do.
ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java