HIVE-29251: Hive ACID: HiveConf object shouldn't be shared between multiple cleanup tasks

deniskuzZ · deniskuzZ · commit ae3be81146b4 · 2025-10-07T21:59:10.000+03:00
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/handler/AbortedTxnCleaner.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/handler/AbortedTxnCleaner.java
@@ -41,6 +41,8 @@
 import java.util.stream.Collectors;
 
 import static java.util.Objects.isNull;
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_COMPACTOR_ABORTEDTXN_THRESHOLD;
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_COMPACTOR_ABORTEDTXN_TIME_THRESHOLD;
 
 /**
  * Abort-cleanup based implementation of TaskHandler.
@@ -51,8 +53,7 @@ class AbortedTxnCleaner extends TaskHandler {
   private static final Logger LOG = LoggerFactory.getLogger(AbortedTxnCleaner.class.getName());
 
   public AbortedTxnCleaner(HiveConf conf, TxnStore txnHandler,
-                           MetadataCache metadataCache, boolean metricsEnabled,
-                           FSRemover fsRemover) {
+        MetadataCache metadataCache, boolean metricsEnabled, FSRemover fsRemover) {
     super(conf, txnHandler, metadataCache, metricsEnabled, fsRemover);
   }
 
@@ -74,17 +75,17 @@ a. Find the list of entries which are suitable for cleanup (This is done in {@li
    **/
   @Override
   public List<Runnable> getTasks() throws MetaException {
-    int abortedThreshold = HiveConf.getIntVar(conf,
-              HiveConf.ConfVars.HIVE_COMPACTOR_ABORTEDTXN_THRESHOLD);
-    long abortedTimeThreshold = HiveConf
-              .getTimeVar(conf, HiveConf.ConfVars.HIVE_COMPACTOR_ABORTEDTXN_TIME_THRESHOLD,
-                      TimeUnit.MILLISECONDS);
-    List<CompactionInfo> readyToCleanAborts = txnHandler.findReadyToCleanAborts(abortedTimeThreshold, abortedThreshold);
+    int abortedThreshold = HiveConf.getIntVar(getConf(), HIVE_COMPACTOR_ABORTEDTXN_THRESHOLD);
+    long abortedTimeThreshold = HiveConf.getTimeVar(getConf(), HIVE_COMPACTOR_ABORTEDTXN_TIME_THRESHOLD,
+        TimeUnit.MILLISECONDS);
+    List<CompactionInfo> readyToCleanAborts =
+        txnHandler.findReadyToCleanAborts(abortedTimeThreshold, abortedThreshold);
 
     if (!readyToCleanAborts.isEmpty()) {
-      return readyToCleanAborts.stream().map(info -> ThrowingRunnable.unchecked(() ->
-                      clean(info, info.minOpenWriteTxnId, metricsEnabled)))
-              .collect(Collectors.toList());
+      return readyToCleanAborts.stream()
+          .map(info -> ThrowingRunnable.unchecked(
+              () -> clean(info, info.minOpenWriteTxnId, metricsEnabled)))
+          .collect(Collectors.toList());
     }
     return Collections.emptyList();
   }
@@ -98,7 +99,9 @@ private void clean(CompactionInfo info, long minOpenWriteTxn, boolean metricsEna
         perfLogger.perfLogBegin(AbortedTxnCleaner.class.getName(), cleanerMetric);
       }
       Partition p = null;
-      Table t = metadataCache.computeIfAbsent(info.getFullTableName(), () -> resolveTable(info.dbname, info.tableName));
+      Table t = metadataCache.computeIfAbsent(info.getFullTableName(),
+          () -> resolveTable(info.dbname, info.tableName));
+
       if (isNull(t)) {
         // The table was dropped before we got around to cleaning it.
         LOG.info("Unable to find table {}, assuming it was dropped.", info.getFullTableName());
@@ -109,25 +112,24 @@ private void clean(CompactionInfo info, long minOpenWriteTxn, boolean metricsEna
         p = resolvePartition(info.dbname, info.tableName, info.partName);
         if (isNull(p)) {
           // The partition was dropped before we got around to cleaning it.
-          LOG.info("Unable to find partition {}, assuming it was dropped.",
-                  info.getFullPartitionName());
+          LOG.info("Unable to find partition {}, assuming it was dropped.", info.getFullPartitionName());
           txnHandler.markCleaned(info);
           return;
         }
       }
 
-      String location = CompactorUtil.resolveStorageDescriptor(t,p).getLocation();
-      info.runAs = TxnUtils.findUserToRunAs(location, t, conf);
+      String location = CompactorUtil.resolveStorageDescriptor(t, p).getLocation();
+      info.runAs = TxnUtils.findUserToRunAs(location, t, getConf());
       abortCleanUsingAcidDir(info, location, minOpenWriteTxn);
 
     } catch (InterruptedException e) {
       LOG.error("Caught an interrupted exception when cleaning, unable to complete cleaning of {} due to {}", info,
-              e.getMessage());
+          e.getMessage());
       handleCleanerAttemptFailure(info, e.getMessage());
       throw e;
     } catch (Exception e) {
       LOG.error("Caught exception when cleaning, unable to complete cleaning of {} due to {}", info,
-              e.getMessage());
+          e.getMessage());
       handleCleanerAttemptFailure(info, e.getMessage());
       throw new MetaException(e.getMessage());
     } finally {
@@ -139,19 +141,23 @@ private void clean(CompactionInfo info, long minOpenWriteTxn, boolean metricsEna
 
   private void abortCleanUsingAcidDir(CompactionInfo info, String location, long minOpenWriteTxn) throws Exception {
     ValidTxnList validTxnList =
-            TxnUtils.createValidTxnListForCleaner(txnHandler.getOpenTxns(), minOpenWriteTxn, true);
+        TxnUtils.createValidTxnListForCleaner(txnHandler.getOpenTxns(), minOpenWriteTxn, true);
     //save it so that getAcidState() sees it
-    conf.set(ValidTxnList.VALID_TXNS_KEY, validTxnList.writeToString());
+    getConf().set(ValidTxnList.VALID_TXNS_KEY, validTxnList.writeToString());
 
     ValidReaderWriteIdList validWriteIdList = getValidCleanerWriteIdList(info, validTxnList);
 
     // Set the highestWriteId of the cleanup equal to the min(minOpenWriteId - 1, highWatermark).
     // This is necessary for looking at the complete state of the table till the min open write Id
     // (if there is an open txn on the table) or the highestWatermark.
     // This is used later on while deleting the records in TXN_COMPONENTS table.
-    info.highestWriteId = Math.min(isNull(validWriteIdList.getMinOpenWriteId()) ?
-            Long.MAX_VALUE : validWriteIdList.getMinOpenWriteId() - 1, validWriteIdList.getHighWatermark());
-    Table table = metadataCache.computeIfAbsent(info.getFullTableName(), () -> resolveTable(info.dbname, info.tableName));
+    info.highestWriteId = Math.min(
+        isNull(validWriteIdList.getMinOpenWriteId()) ?
+            Long.MAX_VALUE : validWriteIdList.getMinOpenWriteId() - 1,
+        validWriteIdList.getHighWatermark());
+
+    Table table = metadataCache.computeIfAbsent(info.getFullTableName(),
+        () -> resolveTable(info.dbname, info.tableName));
     LOG.debug("Cleaning based on writeIdList: {}", validWriteIdList);
 
     boolean success = cleanAndVerifyObsoleteDirectories(info, location, validWriteIdList, table);
@@ -160,6 +166,5 @@ private void abortCleanUsingAcidDir(CompactionInfo info, String location, long m
     } else {
       LOG.warn("Leaving aborted entry {} in TXN_COMPONENTS table.", info);
     }
-
   }
 }
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/handler/CompactionCleaner.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/handler/CompactionCleaner.java
@@ -67,17 +67,16 @@ class CompactionCleaner extends TaskHandler {
   private static final Logger LOG = LoggerFactory.getLogger(CompactionCleaner.class.getName());
 
   public CompactionCleaner(HiveConf conf, TxnStore txnHandler,
-                                MetadataCache metadataCache, boolean metricsEnabled,
-                                FSRemover fsRemover) {
+        MetadataCache metadataCache, boolean metricsEnabled, FSRemover fsRemover) {
     super(conf, txnHandler, metadataCache, metricsEnabled, fsRemover);
   }
 
   @Override
   public List<Runnable> getTasks() throws MetaException {
     long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
-    long retentionTime = HiveConf.getBoolVar(conf, HIVE_COMPACTOR_DELAYED_CLEANUP_ENABLED)
-            ? HiveConf.getTimeVar(conf, HIVE_COMPACTOR_CLEANER_RETENTION_TIME, TimeUnit.MILLISECONDS)
-            : 0;
+    long retentionTime = HiveConf.getBoolVar(getConf(), HIVE_COMPACTOR_DELAYED_CLEANUP_ENABLED)
+        ? HiveConf.getTimeVar(getConf(), HIVE_COMPACTOR_CLEANER_RETENTION_TIME, TimeUnit.MILLISECONDS)
+        : 0;
     List<CompactionInfo> readyToClean = txnHandler.findReadyToClean(minOpenTxnId, retentionTime);
     if (!readyToClean.isEmpty()) {
       long minTxnIdSeenOpen = Math.min(minOpenTxnId, txnHandler.findMinTxnIdSeenOpen());
@@ -88,19 +87,18 @@ public List<Runnable> getTasks() throws MetaException {
       // when min_history_level is finally dropped, than every HMS will commit compaction the new way
       // and minTxnIdSeenOpen can be removed and minOpenTxnId can be used instead.
       return readyToClean.stream().map(ci -> {
-        long cleanerWaterMark = (ci.minOpenWriteId >= 0) ? ci.nextTxnId + 1 : minTxnIdSeenOpen;
-        LOG.info("Cleaning based on min open txn id: {}", cleanerWaterMark);
+        long cleanerWaterMark = (ci.minOpenWriteId > 0) ? ci.nextTxnId + 1 : minTxnIdSeenOpen;
         return ThrowingRunnable.unchecked(() -> clean(ci, cleanerWaterMark, metricsEnabled));
       }).collect(Collectors.toList());
     }
     return Collections.emptyList();
   }
 
   private void clean(CompactionInfo ci, long minOpenTxn, boolean metricsEnabled) throws MetaException {
-    LOG.info("Starting cleaning for {}", ci);
+    LOG.info("Starting cleaning for {} based on min open txn id: {}", ci, minOpenTxn);
     PerfLogger perfLogger = PerfLogger.getPerfLogger(false);
     String cleanerMetric = MetricsConstants.COMPACTION_CLEANER_CYCLE + "_" +
-            (!isNull(ci.type) ? ci.type.toString().toLowerCase() : null);
+        (!isNull(ci.type) ? ci.type.toString().toLowerCase() : null);
     try {
       if (metricsEnabled) {
         perfLogger.perfLogBegin(CompactionCleaner.class.getName(), cleanerMetric);
@@ -111,11 +109,13 @@ private void clean(CompactionInfo ci, long minOpenTxn, boolean metricsEnabled) t
       Partition p = null;
 
       if (isNull(location)) {
-        t = metadataCache.computeIfAbsent(ci.getFullTableName(), () -> resolveTable(ci.dbname, ci.tableName));
+        t = metadataCache.computeIfAbsent(ci.getFullTableName(),
+            () -> resolveTable(ci.dbname, ci.tableName));
+
         if (isNull(t)) {
           // The table was dropped before we got around to cleaning it.
           LOG.info("Unable to find table {}, assuming it was dropped. {}", ci.getFullTableName(),
-                  idWatermark(ci));
+              idWatermark(ci));
           txnHandler.markCleaned(ci);
           return;
         }
@@ -129,8 +129,8 @@ private void clean(CompactionInfo ci, long minOpenTxn, boolean metricsEnabled) t
           p = resolvePartition(ci.dbname, ci.tableName, ci.partName);
           if (isNull(p)) {
             // The partition was dropped before we got around to cleaning it.
-            LOG.info("Unable to find partition {}, assuming it was dropped. {}",
-                    ci.getFullPartitionName(), idWatermark(ci));
+            LOG.info("Unable to find partition {}, assuming it was dropped. {}", ci.getFullPartitionName(),
+                idWatermark(ci));
             txnHandler.markCleaned(ci);
             return;
           }
@@ -146,8 +146,8 @@ private void clean(CompactionInfo ci, long minOpenTxn, boolean metricsEnabled) t
 
       if (!isNull(t) || !isNull(ci.partName)) {
         String path = isNull(location)
-                ? CompactorUtil.resolveStorageDescriptor(t, p).getLocation()
-                : location;
+            ? CompactorUtil.resolveStorageDescriptor(t, p).getLocation()
+            : location;
         boolean dropPartition = !isNull(ci.partName) && isNull(p);
 
         //check if partition wasn't re-created
@@ -161,7 +161,7 @@ private void clean(CompactionInfo ci, long minOpenTxn, boolean metricsEnabled) t
       }
     } catch (Exception e) {
       LOG.error("Caught exception when cleaning, unable to complete cleaning of {} due to {}", ci,
-              e.getMessage());
+          e.getMessage());
       if (metricsEnabled) {
         Metrics.getOrCreateCounter(MetricsConstants.COMPACTION_CLEANER_FAILURE_COUNTER).inc();
       }
@@ -205,9 +205,9 @@ private void cleanUsingLocation(CompactionInfo ci, String path, boolean requires
 
   private void cleanUsingAcidDir(CompactionInfo ci, String location, long minOpenTxn) throws Exception {
     ValidTxnList validTxnList =
-            TxnUtils.createValidTxnListForCleaner(txnHandler.getOpenTxns(), minOpenTxn, false);
+        TxnUtils.createValidTxnListForCleaner(txnHandler.getOpenTxns(), minOpenTxn, false);
     //save it so that getAcidState() sees it
-    conf.set(ValidTxnList.VALID_TXNS_KEY, validTxnList.writeToString());
+    getConf().set(ValidTxnList.VALID_TXNS_KEY, validTxnList.writeToString());
     /*
      * {@code validTxnList} is capped by minOpenTxnGLB so if
      * {@link AcidUtils#getAcidState(Path, Configuration, ValidWriteIdList)} sees a base/delta
@@ -241,7 +241,8 @@ private void cleanUsingAcidDir(CompactionInfo ci, String location, long minOpenT
 
     // Creating 'reader' list since we are interested in the set of 'obsolete' files
     ValidReaderWriteIdList validWriteIdList = getValidCleanerWriteIdList(ci, validTxnList);
-    Table table = metadataCache.computeIfAbsent(ci.getFullTableName(), () -> resolveTable(ci.dbname, ci.tableName));
+    Table table = metadataCache.computeIfAbsent(ci.getFullTableName(),
+        () -> resolveTable(ci.dbname, ci.tableName));
     LOG.debug("Cleaning based on writeIdList: {}", validWriteIdList);
 
     boolean success = cleanAndVerifyObsoleteDirectories(ci, location, validWriteIdList, table);
@@ -254,7 +255,8 @@ private void cleanUsingAcidDir(CompactionInfo ci, String location, long minOpenT
   }
 
   private LockRequest createLockRequest(CompactionInfo ci) {
-    return CompactorUtil.createLockRequest(conf, ci, 0, LockType.EXCL_WRITE, DataOperationType.DELETE);
+    return CompactorUtil.createLockRequest(
+        getConf(), ci, 0, LockType.EXCL_WRITE, DataOperationType.DELETE);
   }
 
   private static String idWatermark(CompactionInfo ci) {
@@ -263,7 +265,7 @@ private static String idWatermark(CompactionInfo ci) {
 
   @Override
   protected ValidReaderWriteIdList getValidCleanerWriteIdList(CompactionInfo ci, ValidTxnList validTxnList)
-          throws NoSuchTxnException, MetaException {
+        throws NoSuchTxnException, MetaException {
     ValidReaderWriteIdList validWriteIdList = super.getValidCleanerWriteIdList(ci, validTxnList);
     /*
      * We need to filter the obsoletes dir list, to only remove directories that were made obsolete by this compaction
@@ -279,11 +281,14 @@ protected ValidReaderWriteIdList getValidCleanerWriteIdList(CompactionInfo ci, V
   private CleanupRequest getCleaningRequestBasedOnLocation(CompactionInfo ci, String location) {
     String strIfPurge = ci.getProperty("ifPurge");
     boolean ifPurge = strIfPurge != null || Boolean.parseBoolean(ci.getProperty("ifPurge"));
-
     Path obsoletePath = new Path(location);
+
     return new CleanupRequestBuilder()
-            .setLocation(location).setDbName(ci.dbname).setFullPartitionName(ci.getFullPartitionName())
-            .setRunAs(ci.runAs).setPurge(ifPurge).setObsoleteDirs(Collections.singletonList(obsoletePath))
-            .build();
+        .setLocation(location).setDbName(ci.dbname)
+        .setFullPartitionName(ci.getFullPartitionName())
+        .setRunAs(ci.runAs)
+        .setPurge(ifPurge)
+        .setObsoleteDirs(Collections.singletonList(obsoletePath))
+        .build();
   }
 }
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/handler/TaskHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/handler/TaskHandler.java
