@Test
  public void testAbortAfterMarkCleaned() throws Exception {
    assumeTrue(MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.COMPACTOR_CLEAN_ABORTS_USING_CLEANER));
    String dbName = "default";
    String tableName = "cws";

    String agentInfo = "UT_" + Thread.currentThread().getName();

    dropTables(tableName);
    executeStatementOnDriver("CREATE TABLE " + tableName + "(a STRING, b STRING) " + //currently ACID requires table to be bucketed
            " STORED AS ORC  TBLPROPERTIES ('transactional'='true')", driver);
    executeStatementOnDriver("insert into table " + tableName + " values ('1', '2'), ('3', '4') ", driver);
    executeStatementOnDriver("insert into table " + tableName + " values ('1', '2'), ('3', '4') ", driver);


    StrictDelimitedInputWriter writer = StrictDelimitedInputWriter.newBuilder()
            .withFieldDelimiter(',')
            .build();

    // Create three folders with two different transactions
    HiveStreamingConnection connection1 = HiveStreamingConnection.newBuilder()
            .withDatabase(dbName)
            .withTable(tableName)
            .withAgentInfo(agentInfo)
            .withHiveConf(conf)
            .withRecordWriter(writer)
            .withStreamingOptimizations(true)
            .withTransactionBatchSize(1)
            .connect();

    HiveStreamingConnection connection2 = HiveStreamingConnection.newBuilder()
            .withDatabase(dbName)
            .withTable(tableName)
            .withAgentInfo(agentInfo)
            .withHiveConf(conf)
            .withRecordWriter(writer)
            .withStreamingOptimizations(true)
            .withTransactionBatchSize(1)
            .connect();

    // Abort a transaction which writes data.
    connection1.beginTransaction();
    connection1.write("1,1".getBytes());
    connection1.write("2,1".getBytes());
    connection1.abortTransaction();

    // Open a txn which is opened and long running.
    connection2.beginTransaction();
    connection2.write("3,1".getBytes());

    Cleaner cleaner = new Cleaner();
    TxnStore mockedTxnHandler = Mockito.spy(TxnUtils.getTxnStore(conf));
    doAnswer(invocationOnMock -> {
      connection2.abortTransaction();
      return invocationOnMock.callRealMethod();
    }).when(mockedTxnHandler).markCleaned(any());

    MetadataCache metadataCache = new MetadataCache(false);
    FSRemover fsRemover = new FSRemover(conf, ReplChangeManager.getInstance(conf), metadataCache);
    cleaner.setConf(conf);
    List<TaskHandler> cleanupHandlers = TaskHandlerFactory.getInstance()
            .getHandlers(conf, mockedTxnHandler, metadataCache, false, fsRemover);
    cleaner.init(new AtomicBoolean(true));
    cleaner.setCleanupHandlers(cleanupHandlers);
    cleaner.run();

    int count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
    Assert.assertEquals(TestTxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 1, count);
  }
After each major compaction, stats need to be updated on the table
1. create an ORC backed table (Orc is currently required by ACID)
2. populate with data
3. compute stats
4. Trigger major compaction (which should update stats)
5. check that stats have been updated

@throws Exception todo:
                  4. add a test with sorted table?
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java