HIVE-29571: ACID Compaction: Marking the compaction as compacted should happen after its txn got committed

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/service/AcidCompactionService.java

@@ -48,6 +48,7 @@
 import org.apache.hadoop.hive.ql.txn.compactor.CompactorFactory;
 import org.apache.hadoop.hive.ql.txn.compactor.CompactorPipeline;
 import org.apache.hadoop.hive.ql.txn.compactor.CompactorUtil;
+import org.apache.hadoop.hive.ql.txn.compactor.CompactorUtil.ThrowingRunnable;
 import org.apache.hadoop.hive.ql.txn.compactor.QueryCompactor;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hive.common.util.Ref;
@@ -211,32 +212,27 @@ public Boolean compact(Table table, CompactionInfo ci) throws Exception {
 
       // Don't start compaction or cleaning if not necessary
       if (isDynPartAbort(table, ci)) {
-        msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci));
-        compactionTxn.wasSuccessful();
+        compactionTxn.markForCommit(() -> msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci)));
         return false;
       }
       dir = getAcidStateForWorker(ci, sd, tblValidWriteIds);
       if (!isEnoughToCompact(ci, dir, sd)) {
         if (needsCleaning(dir, sd)) {
-          msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci));
+          compactionTxn.markForCommit(() -> msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci)));
         } else {
           // do nothing
           ci.errorMessage = "None of the compaction thresholds met, compaction request is refused!";
           LOG.debug(ci.errorMessage + " Compaction info: {}", ci);
-          msc.markRefused(CompactionInfo.compactionInfoToStruct(ci));
+          compactionTxn.markForCommit(() -> msc.markRefused(CompactionInfo.compactionInfoToStruct(ci)));
+
         }
-        compactionTxn.wasSuccessful();
         return false;
       }
       if (!ci.isMajorCompaction() && !CompactorUtil.isMinorCompactionSupported(conf, table.getParameters(), dir)) {
         ci.errorMessage = "Query based Minor compaction is not possible for full acid tables having raw format " +
             "(non-acid) data in them.";
         LOG.error(ci.errorMessage + " Compaction info: {}", ci);
-        try {
-          msc.markRefused(CompactionInfo.compactionInfoToStruct(ci));
-        } catch (Throwable tr) {
-          LOG.error("Caught an exception while trying to mark compaction {} as failed: {}", ci, tr);
-        }
+        compactionTxn.markForAbort(() -> msc.markRefused(CompactionInfo.compactionInfoToStruct(ci)));
         return false;
       }
       CompactorUtil.checkInterrupt(CLASS_NAME);
@@ -261,8 +257,7 @@ public Boolean compact(Table table, CompactionInfo ci) throws Exception {
 
         LOG.info("Completed " + ci.type.toString() + " compaction for " + ci.getFullPartitionName() + " in "
             + compactionTxn + ", marking as compacted.");
-        msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci));
-        compactionTxn.wasSuccessful();
+        compactionTxn.markForCommit(() -> msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci)));
 
         AcidMetricService.updateMetricsFromWorker(ci.dbname, ci.tableName, ci.partName, ci.type,
             dir.getCurrentDirectories().size(), dir.getDeleteDeltas().size(), conf, msc);
@@ -346,7 +341,11 @@ class CompactionTxn implements AutoCloseable {
     private long lockId = 0;
 
     private TxnStatus status = TxnStatus.UNKNOWN;
-    private boolean successfulCompaction = false;
+
+    private ThrowingRunnable onCommitSuccess;
+    private ThrowingRunnable onAbortSuccess;
+
+    private boolean rollbackOnly = true;
 
     /**
      * Try to open a new txn.
@@ -377,11 +376,14 @@ private LockRequest createLockRequest(CompactionInfo ci) {
       return CompactorUtil.createLockRequest(conf, ci, txnId, lockAndOpType.getKey(), lockAndOpType.getValue());
     }
 
-    /**
-     * Mark compaction as successful. This means the txn will be committed; otherwise it will be aborted.
-     */
-    void wasSuccessful() {
-      this.successfulCompaction = true;
+    void markForCommit(ThrowingRunnable action) {
+      this.rollbackOnly = false;
+      this.onCommitSuccess = action;
+    }
+
+    void markForAbort(ThrowingRunnable action) {
+      this.rollbackOnly = true;
+      this.onAbortSuccess = action;
     }
 
     /**
@@ -396,10 +398,16 @@ public void close() throws Exception {
         //the transaction is about to close, we can stop heartbeating regardless of it's state
         CompactionHeartbeatService.getInstance(conf).stopHeartbeat(txnId);
       } finally {
-        if (successfulCompaction) {
-          commit();
-        } else {
+        if (rollbackOnly) {
           abort();
+          if (onAbortSuccess != null) {
+            onAbortSuccess.run();
+          }
+          return;
+        }
+        commit();
+        if (onCommitSuccess != null) {
+          onCommitSuccess.run();
         }
       }
     }