@Test
  public void cleanupAfterMajorTableCompactionWithLongRunningQuery() throws Exception {
    Table t = newTable("default", "camtc", false);

    addBaseFile(t, null, 20L, 20);
    addDeltaFile(t, null, 21L, 22L, 2);
    addDeltaFile(t, null, 23L, 24L, 2);
    addBaseFile(t, null, 25L, 25, 26);

    burnThroughTransactions("default", "camtc", 25);

    CompactionRequest rqst = new CompactionRequest("default", "camtc", CompactionType.MAJOR);
    txnHandler.compact(rqst);

    FindNextCompactRequest findNextCompactRequest = new FindNextCompactRequest();
    findNextCompactRequest.setWorkerId("fred");
    findNextCompactRequest.setWorkerVersion(WORKER_VERSION);
    CompactionInfo ci = txnHandler.findNextToCompact(findNextCompactRequest);
    ci.runAs = System.getProperty("user.name");
    long compactTxn = openTxn(TxnType.COMPACTION);

    ValidTxnList validTxnList = TxnCommonUtils.createValidReadTxnList(
        txnHandler.getOpenTxns(Collections.singletonList(TxnType.READ_ONLY)), compactTxn);
    GetValidWriteIdsRequest validWriteIdsRqst = new GetValidWriteIdsRequest(Collections.singletonList(ci.getFullTableName()));
    validWriteIdsRqst.setValidTxnList(validTxnList.writeToString());

    ValidCompactorWriteIdList tblValidWriteIds = TxnUtils.createValidCompactWriteIdList(
        txnHandler.getValidWriteIds(validWriteIdsRqst).getTblValidWriteIds().get(0));
    ci.highestWriteId = tblValidWriteIds.getHighWatermark();
    txnHandler.updateCompactorState(ci, compactTxn);

    txnHandler.markCompacted(ci);
    // Open a query during compaction
    long longQuery = openTxn();
    if (useMinHistoryWriteId()) {
      allocateTableWriteId("default", "camtc", longQuery);
    }
    txnHandler.commitTxn(new CommitTxnRequest(compactTxn));

    startCleaner();

    // The long running query should prevent the cleanup
    ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
    Assert.assertEquals(1, rsp.getCompactsSize());
    Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());

    // Check that the files are not removed
    List<Path> paths = getDirectories(conf, t, null);
    Assert.assertEquals(4, paths.size());

    // After the commit cleaning can proceed
    txnHandler.commitTxn(new CommitTxnRequest(longQuery));
    Thread.sleep(MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.TXN_OPENTXN_TIMEOUT, TimeUnit.MILLISECONDS));
    startCleaner();

    rsp = txnHandler.showCompact(new ShowCompactRequest());
    Assert.assertEquals(1, rsp.getCompactsSize());
    Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, rsp.getCompacts().get(0).getState());

    // Check that the files are removed
    paths = getDirectories(conf, t, null);
    Assert.assertEquals(1, paths.size());
    Assert.assertEquals(addVisibilitySuffix("base_25", 26), paths.get(0).getName());
  }
Tests for the compactor Cleaner thread
ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java