This is an automated email from the ASF dual-hosted git repository.
dengzhhu653 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 295a0e79169 HIVE-30158: Run the housekeeping tasks always under
dynamic election (#6868)
295a0e79169 is described below
commit 295a0e79169c1eee7f2b2746526b7ab1f723d958
Author: dengzh <[email protected]>
AuthorDate: Tue Oct 20 08:35:18 2026 +0800
HIVE-30158: Run the housekeeping tasks always under dynamic election (#6868)
---
.../hadoop/hive/metastore/conf/MetastoreConf.java | 25 ++--
.../hive/metastore/leader/CompactorTasks.java | 37 +++++-
.../hive/metastore/leader/HouseKeepingTasks.java | 18 ++-
.../hive/metastore/leader/LeaderElection.java | 9 +-
.../hive/metastore/leader/LeaseLeaderElection.java | 5 +
.../leader/TestDynamicLeaderHousekeeping.java | 143 +++++++++++++++++++++
6 files changed, 212 insertions(+), 25 deletions(-)
diff --git
a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
index 733bf0dec42..c36fc205d98 100644
---
a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
+++
b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
@@ -625,10 +625,11 @@ public enum ConfVars {
"The database where the Hive lock sits when
metastore.housekeeping.leader.election is set to lock."),
METASTORE_HOUSEKEEPING_THREADS_ON("metastore.housekeeping.threads.on",
"hive.metastore.housekeeping.threads.on", true,
- "Whether to run the tasks under metastore.task.threads.remote on this
metastore instance or not.\n" +
- "Set this to true on one instance of the Thrift metastore service
as part of turning\n" +
- "on Hive transactions. For a complete list of parameters required
for turning on\n" +
- "transactions, see hive.txn.manager."),
+ "Whether to run the tasks under metastore.task.threads.remote on this
metastore instance when\n" +
+ "metastore.housekeeping.leader.election is host. Ignored when
leader election is lock: the\n" +
+ "elected leader always runs those tasks. To disable a specific
task, set its run interval\n" +
+ "to 0 (for example metastore.partition.management.task.frequency).
For a complete list of\n" +
+ "parameters required for turning on transactions, see
hive.txn.manager."),
METASTORE_ACIDMETRICS_THREAD_ON("metastore.acidmetrics.thread.on",
"hive.metastore.acidmetrics.thread.on", true,
"Whether to run acid related metrics collection on this metastore
instance."),
@@ -670,15 +671,15 @@ public enum ConfVars {
"Thrift API method names for which to record the per-query metrics.\n"
+
"Per-query slow detection applies to all JDBC executions while
profiling is on."),
COMPACTOR_INITIATOR_ON("metastore.compactor.initiator.on",
"hive.compactor.initiator.on", true,
- "Whether to run the initiator thread on this metastore instance or
not.\n" +
- "Set this to true on one instance of the Thrift metastore service
as part of turning\n" +
- "on Hive transactions. For a complete list of parameters required
for turning on\n" +
- "transactions, see hive.txn.manager."),
+ "Whether to run the initiator thread on this metastore instance
when\n" +
+ "metastore.housekeeping.leader.election is host. Ignored when
leader election is lock:\n" +
+ "the elected leader always runs the initiator. For a complete list
of parameters required\n" +
+ "for turning on transactions, see hive.txn.manager."),
COMPACTOR_CLEANER_ON("metastore.compactor.cleaner.on",
"hive.compactor.cleaner.on", true,
- "Whether to run the cleaner thread on this metastore instance or
not.\n" +
- "Set this to true on one instance of the Thrift metastore service
as part of turning\n" +
- "on Hive transactions. For a complete list of parameters required
for turning on\n" +
- "transactions, see hive.txn.manager."),
+ "Whether to run the cleaner thread on this metastore instance when\n" +
+ "metastore.housekeeping.leader.election is host. Ignored when
leader election is lock:\n" +
+ "the elected leader always runs the cleaner if acid is supported.
For a complete list of parameters required\n" +
+ "for turning on transactions, see hive.txn.manager."),
COMPACTOR_INITIATOR_TABLECACHE_ON("metastore.compactor.initiator.tablecache.on",
"hive.compactor.initiator.tablecache.on", true,
"Enable table caching in the initiator. Currently the cache is cleaned
after each cycle."),
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/CompactorTasks.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/CompactorTasks.java
index 0b4a247d40e..cecc7ccaaef 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/CompactorTasks.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/CompactorTasks.java
@@ -19,6 +19,8 @@
package org.apache.hadoop.hive.metastore.leader;
+import com.google.common.annotations.VisibleForTesting;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.metastore.HiveMetaStore;
import org.apache.hadoop.hive.metastore.MetaStoreThread;
@@ -41,6 +43,8 @@ public class CompactorTasks implements
LeaderElection.LeadershipStateListener {
// each MetaStoreThread runs as a thread
private Map<MetaStoreThread, AtomicBoolean> metastoreThreadsMap;
+ private volatile boolean isDynamic;
+
public CompactorTasks(Configuration configuration) {
// recreate a new configuration
this.configuration = new Configuration(requireNonNull(configuration,
@@ -59,13 +63,23 @@ private MetaStoreThread instantiateThread(String classname)
throws Exception {
}
}
+ private boolean shouldScheduleInitiator() {
+ return isDynamic
+ || MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON);
+ }
+
+ private boolean shouldScheduleCleaner() {
+ return isDynamic
+ || MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON);
+ }
+
public List<MetaStoreThread> getCompactorThreads() throws Exception {
List<MetaStoreThread> compactors = new ArrayList<>();
- if (MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON)) {
+ if (shouldScheduleInitiator()) {
MetaStoreThread initiator =
instantiateThread("org.apache.hadoop.hive.ql.txn.compactor.Initiator");
compactors.add(initiator);
}
- if (MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON)) {
+ if (shouldScheduleCleaner()) {
if (MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID)) {
MetaStoreThread cleaner =
instantiateThread("org.apache.hadoop.hive.ql.txn.compactor.Cleaner");
compactors.add(cleaner);
@@ -93,14 +107,18 @@ private void logCompactionParameters() {
MetastoreConf.getIntVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_HISTORY_RETENTION_SUCCEEDED));
HiveMetaStore.LOG.info("metastore.compactor.initiator.failed.compacts.threshold
= {}",
MetastoreConf.getIntVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_INITIATOR_FAILED_THRESHOLD));
- HiveMetaStore.LOG.info("metastore.compactor.enable.stats.compression",
+ HiveMetaStore.LOG.info("metastore.compactor.enable.stats.compression = {}",
MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_MINOR_STATS_COMPRESSION));
if (!MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON)) {
- HiveMetaStore.LOG.warn("Compactor Initiator is turned Off. Automatic
compaction will not be triggered.");
+ String message = isDynamic ? "metastore.compactor.initiator.on is
ignored since current leader is dynamic." :
+ "Compactor Initiator is turned Off. Automatic compaction will not be
triggered.";
+ HiveMetaStore.LOG.warn(message);
}
if (!MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON)) {
- HiveMetaStore.LOG.warn("Compactor Cleaner is turned Off. Automatic
compaction cleaner will not be triggered.");
+ String message = isDynamic ? "metastore.compactor.cleaner.on is ignored
since current leader is dynamic." :
+ "Compactor Cleaner is turned Off. Automatic compaction cleaner will
not be triggered.";
+ HiveMetaStore.LOG.warn(message);
}
}
@@ -110,6 +128,7 @@ public void takeLeadership(LeaderElection election) throws
Exception {
throw new IllegalStateException("There should be no running tasks before
taking the leadership!");
}
+ isDynamic = election.isDynamic();
logCompactionParameters();
metastoreThreadsMap = new IdentityHashMap<>();
List<MetaStoreThread> metaStoreThreads = getCompactorThreads();
@@ -119,7 +138,7 @@ public void takeLeadership(LeaderElection election) throws
Exception {
thread.init(flag);
thread.enforceMutex(election.enforceMutex());
metastoreThreadsMap.put(thread, flag);
- HiveMetaStore.LOG.info("Starting metastore thread of type " +
thread.getClass().getName());
+ HiveMetaStore.LOG.info("Starting metastore thread: {}",
thread.getClass().getName());
thread.start();
}
}
@@ -132,10 +151,14 @@ public void lossLeadership(LeaderElection election) {
if (thread instanceof Thread) {
((Thread)thread).interrupt();
}
- HiveMetaStore.LOG.info("Stopped the Compaction task: {}.",
thread.getClass().getName());
+ HiveMetaStore.LOG.info("Stopped the metastore thread: {}.",
thread.getClass().getName());
});
metastoreThreadsMap = null;
}
}
+ @VisibleForTesting
+ public Map<MetaStoreThread, AtomicBoolean> getMetastoreThreadsMap() {
+ return metastoreThreadsMap;
+ }
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/HouseKeepingTasks.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/HouseKeepingTasks.java
index f2e69bb92ea..897fb0a85df 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/HouseKeepingTasks.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/HouseKeepingTasks.java
@@ -49,6 +49,8 @@ public class HouseKeepingTasks implements
LeaderElection.LeadershipStateListener
private List<MetastoreTaskThread> runningTasks;
+ private volatile boolean isDynamic;
+
public HouseKeepingTasks(Configuration configuration, boolean
runOnlyRemoteTasks) {
this.configuration = new Configuration(requireNonNull(configuration,
"configuration is null"));
@@ -60,12 +62,18 @@ public HouseKeepingTasks(Configuration configuration,
boolean runOnlyRemoteTasks
*/
public List<MetastoreTaskThread> getRemoteOnlyTasks() throws Exception {
List<MetastoreTaskThread> remoteOnlyTasks = new ArrayList<>();
- if(!MetastoreConf.getBoolVar(configuration,
- MetastoreConf.ConfVars.METASTORE_HOUSEKEEPING_THREADS_ON)) {
+ boolean tasksOn =
+ MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.METASTORE_HOUSEKEEPING_THREADS_ON);
+ if(!isDynamic && !tasksOn) {
return remoteOnlyTasks;
}
- boolean isCompactorEnabled = MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON)
- || MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON);
+ if (!tasksOn) {
+ HiveMetaStore.LOG.info("metastore.housekeeping.threads.on is ignored
since current leader is dynamic.");
+ }
+
+ boolean isCompactorEnabled = isDynamic
+ || MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON)
+ || MetastoreConf.getBoolVar(configuration,
MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON);
Collection<String> taskNames =
MetastoreConf.getStringCollection(configuration,
MetastoreConf.ConfVars.TASK_THREADS_REMOTE_ONLY);
@@ -80,7 +88,6 @@ public List<MetastoreTaskThread> getRemoteOnlyTasks() throws
Exception {
return remoteOnlyTasks;
}
- // Copied from HiveMetaStore
public List<MetastoreTaskThread> getAlwaysTasks() throws Exception {
List<MetastoreTaskThread> alwaysTasks = new ArrayList<>();
Collection<String> taskNames =
@@ -102,6 +109,7 @@ public void takeLeadership(LeaderElection election) throws
Exception {
ThreadFactory threadFactory = new ThreadFactoryBuilder().setDaemon(true)
.setNameFormat("Metastore Scheduled Worker(" + election.getName() + ")
%d").build();
final List<MetastoreTaskThread> tasks;
+ isDynamic = election.isDynamic();
if (!runOnlyRemoteTasks) {
tasks = new ArrayList<>(getAlwaysTasks());
} else {
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaderElection.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaderElection.java
index 24d824d71a2..f491b306804 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaderElection.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaderElection.java
@@ -75,6 +75,14 @@ default boolean enforceMutex() {
return true;
}
+ /**
+ * Whether this leader can be changed during selection.
+ * @return true if the leader can be voted again, false otherwise.
+ */
+ default boolean isDynamic() {
+ return false;
+ }
+
interface LeadershipStateListener {
/**
@@ -90,5 +98,4 @@ interface LeadershipStateListener {
void lossLeadership(LeaderElection election) throws Exception;
}
-
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaseLeaderElection.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaseLeaderElection.java
index 187ceb5f589..b3bbbbd8e47 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaseLeaderElection.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaseLeaderElection.java
@@ -433,4 +433,9 @@ public String getName() {
public boolean enforceMutex() {
return this.enforceMutex;
}
+
+ @Override
+ public boolean isDynamic() {
+ return true;
+ }
}
diff --git
a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/leader/TestDynamicLeaderHousekeeping.java
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/leader/TestDynamicLeaderHousekeeping.java
new file mode 100644
index 00000000000..460e8f532b9
--- /dev/null
+++
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/leader/TestDynamicLeaderHousekeeping.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hadoop.hive.metastore.leader;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.MetaStoreThread;
+import org.apache.hadoop.hive.metastore.MetastoreTaskThread;
+import org.apache.hadoop.hive.metastore.annotation.MetastoreUnitTest;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import
org.apache.hadoop.hive.metastore.txn.service.CompactionHouseKeeperService;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@Category(MetastoreUnitTest.class)
+public class TestDynamicLeaderHousekeeping {
+
+ private HouseKeepingTasks houseKeepingTasks;
+ private CompactorTasks compactorTasks;
+ private static LeaseLeaderElection dynamicElection;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ dynamicElection = new LeaseLeaderElection();
+ dynamicElection.setName("housekeeping test");
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (houseKeepingTasks != null) {
+ houseKeepingTasks.lossLeadership(dynamicElection);
+ houseKeepingTasks = null;
+ }
+ compactorTasks = null;
+ }
+
+ @Test
+ public void remoteTasksSkippedOnStaticLeader() throws Exception {
+ Configuration conf = baseConf();
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.METASTORE_HOUSEKEEPING_THREADS_ON, false);
+ MetastoreConf.setVar(conf, MetastoreConf.ConfVars.TASK_THREADS_REMOTE_ONLY,
+ NoOpMetastoreTaskThread.class.getName());
+
+ houseKeepingTasks = new HouseKeepingTasks(conf, true);
+ Assert.assertTrue(houseKeepingTasks.getRemoteOnlyTasks().isEmpty());
+ }
+
+ @Test
+ public void remoteTasksScheduledOnDynamicLeader() throws Exception {
+ Configuration conf = baseConf();
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.METASTORE_HOUSEKEEPING_THREADS_ON, false);
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON, false);
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON, false);
+ MetastoreConf.setVar(conf, MetastoreConf.ConfVars.TASK_THREADS_REMOTE_ONLY,
+ NoOpMetastoreTaskThread.class.getName() + "," +
CompactionHouseKeeperService.class.getName());
+ MetastoreConf.setLongVar(conf, MetastoreConf.ConfVars.THREAD_POOL_SIZE, 1);
+
+ houseKeepingTasks = new HouseKeepingTasks(conf, true);
+ StaticLeaderElection staticElection = new StaticLeaderElection();
+ houseKeepingTasks.takeLeadership(staticElection);
+ List<MetastoreTaskThread> remoteOnlyTasks =
houseKeepingTasks.getRemoteOnlyTasks();
+ Assert.assertTrue(remoteOnlyTasks.isEmpty());
+ Assert.assertFalse(houseKeepingTasks.getExecutorService().isShutdown());
+ houseKeepingTasks.lossLeadership(staticElection);
+ Assert.assertNull(houseKeepingTasks.getExecutorService());
+
+ houseKeepingTasks.takeLeadership(dynamicElection);
+ remoteOnlyTasks = houseKeepingTasks.getRemoteOnlyTasks();
+ Assert.assertEquals(2, remoteOnlyTasks.size());
+ Assert.assertFalse(houseKeepingTasks.getExecutorService().isShutdown());
+ }
+
+ @Test
+ public void compactorOnDynamicLeader() throws Exception {
+ Configuration conf = baseConf();
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON, false);
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON, false);
+
+ compactorTasks = new CompactorTasks(conf);
+ StaticLeaderElection staticElection = new StaticLeaderElection();
+ compactorTasks.takeLeadership(staticElection);
+ Map<MetaStoreThread, AtomicBoolean> metastoreThreadsMap =
compactorTasks.getMetastoreThreadsMap();
+ Assert.assertTrue(metastoreThreadsMap.isEmpty());
+ compactorTasks.lossLeadership(staticElection);
+ metastoreThreadsMap = compactorTasks.getMetastoreThreadsMap();
+ Assert.assertNull(metastoreThreadsMap);
+
+ ClassNotFoundException cnfe =
Assert.assertThrows(ClassNotFoundException.class,
+ () -> compactorTasks.takeLeadership(dynamicElection));
+ // Error in starting the Initiator
+
Assert.assertTrue(cnfe.getMessage().contains("org.apache.hadoop.hive.ql.txn.compactor.Initiator"));
+ }
+
+ private static Configuration baseConf() {
+ Configuration conf = MetastoreConf.newMetastoreConf();
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.METASTORE_SUPPORT_ACID, true);
+ return conf;
+ }
+
+ public static class NoOpMetastoreTaskThread implements MetastoreTaskThread {
+ @Override
+ public void setConf(Configuration conf) {
+ }
+
+ @Override
+ public Configuration getConf() {
+ return null;
+ }
+
+ @Override
+ public void run() {
+ }
+
+ @Override
+ public long runFrequency(TimeUnit unit) {
+ return unit.convert(1, TimeUnit.HOURS);
+ }
+ }
+}
