@Test
  public void processCompactionCandidatesInParallel() throws Exception {
    Table t = newTable("default", "camipc", true);
    List<Partition> partitions = new ArrayList<>();
    Partition p;
    for (int i = 0; i < 10; i++) {
      p = newPartition(t, "today" + i);

      addBaseFile(t, p, 20L, 20);
      addDeltaFile(t, p, 21L, 22L, 2);
      addDeltaFile(t, p, 23L, 24L, 2);
      addDeltaFile(t, p, 21L, 24L, 4);
      partitions.add(p);
    }

    burnThroughTransactions("default", "camipc", 25);
    for (int i = 0; i < 10; i++) {
      CompactionRequest rqst = new CompactionRequest("default", "camipc", CompactionType.MINOR);
      rqst.setPartitionname("ds=today" + i);
      compactInTxn(rqst);
    }

    conf.setIntVar(HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_THREADS_NUM, 3);
    startCleaner();

    // Check there are no compactions requests left.
    ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
    Assert.assertEquals(10, rsp.getCompactsSize());
    Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, rsp.getCompacts().get(0).getState());

    // Check that the files are removed
    for (Partition pa : partitions) {
      List<Path> paths = getDirectories(conf, t, pa);
      Assert.assertEquals(2, paths.size());
      boolean sawBase = false, sawDelta = false;
      for (Path path : paths) {
        if (path.getName().equals("base_20")) {
          sawBase = true;
        } else if (path.getName().equals(makeDeltaDirNameCompacted(21, 24))) {
          sawDelta = true;
        } else {
          Assert.fail("Unexpected file " + path.getName());
        }
      }
      Assert.assertTrue(sawBase);
      Assert.assertTrue(sawDelta);
    }
  }
Tests for the compactor Cleaner thread
ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java