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 18628fde9f5 HIVE-29720: Abort the service if Metastore cannot
participate in the election (#6621)
18628fde9f5 is described below
commit 18628fde9f53592d3dad924ef215e8513a4337b0
Author: Arun <[email protected]>
AuthorDate: Fri Jul 31 05:28:41 2026 +0530
HIVE-29720: Abort the service if Metastore cannot participate in the
election (#6621)
---
.../hadoop/hive/metastore/HiveMetaStore.java | 3 +-
.../metastore/leader/LeaderElectionContext.java | 93 ++++++++++---
.../hive/metastore/leader/TestLeaderElection.java | 150 +++++++++++++++++++++
3 files changed, 224 insertions(+), 22 deletions(-)
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
index b7a0004c76b..d6a93a75050 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
@@ -869,7 +869,8 @@ public void run() {
}
context.start();
} catch (Throwable e) {
- LOG.error("Failure when starting the leader tasks, Compaction or
Housekeeping tasks may not happen", e);
+ LOG.error("Failed to initialize Metastore leader elections;
Metastore will start without leader-only tasks "
+ + "(compaction, housekeeping). Runtime election failures will
abort the Metastore instead.", e);
} finally {
startLock.unlock();
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaderElectionContext.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaderElectionContext.java
index 14d18e832db..01bc44c9282 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaderElectionContext.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/leader/LeaderElectionContext.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hive.metastore.leader;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.common.TableName;
@@ -27,12 +28,14 @@
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import
org.apache.hadoop.hive.metastore.leader.LeaderElection.LeadershipStateListener;
import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
+import org.apache.hadoop.util.ExitUtil;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
import static java.util.Objects.requireNonNull;
@@ -66,24 +69,27 @@ public String getName() {
private final Configuration conf;
private final String servHost;
- // Whether the context should be started as a daemon
- private final boolean startAsDaemon;
// Audit the event of election
private AuditLeaderListener auditLeaderListener;
// State change listeners group by type
private final Map<TTYPE, List<LeadershipStateListener>> listeners;
// Collection of leader candidates
private final List<LeaderElection<?>> leaderElections = new ArrayList<>();
- // Property for testing, a single leader will be created
+ // Ensures the abort action runs at most once even if both electors fail
concurrently.
+ private final AtomicBoolean isAborting = new AtomicBoolean(false);
+ // Action to run when a leader election exhausts retries. Defaults to
ExitUtil.terminate,
+ // which fires the HMS shutdown-hook chain (election close, ZK
deregistration, metrics,
+ // servlet server). ExitUtil is preferred over raw System.exit for JDK 21
compatibility
+ // and test-time exit interception. Overridable for tests.
+ private Runnable abortAction = () -> ExitUtil.terminate(1);
private LeaderElectionContext(String servHost, Configuration conf,
Map<TTYPE, List<LeadershipStateListener>> listeners,
- boolean startAsDaemon, IHMSHandler handler) throws Exception {
+ IHMSHandler handler) throws Exception {
requireNonNull(conf, "conf is null");
requireNonNull(listeners, "listeners is null");
this.servHost = servHost;
this.conf = new Configuration(conf);
- this.startAsDaemon = startAsDaemon;
String tableName = MetastoreConf.getVar(conf,
MetastoreConf.ConfVars.METASTORE_HOUSEKEEPING_LEADER_AUDITTABLE);
if (StringUtils.isNotEmpty(tableName)) {
@@ -97,6 +103,7 @@ private LeaderElectionContext(String servHost, Configuration
conf,
public void start() throws Exception {
List<TTYPE> ttypes = new ArrayList<>(listeners.keySet());
Collections.shuffle(ttypes);
+ List<Thread> daemons = new ArrayList<>();
for (TTYPE ttype : ttypes) {
List<LeadershipStateListener> listenerList = listeners.get(ttype);
if (listenerList.isEmpty()) {
@@ -117,19 +124,24 @@ public void start() throws Exception {
throw new RuntimeException("Error claiming to be leader: " +
leaderElection.getName(), e);
}
});
-
- if (startAsDaemon) {
- daemon.setName("Metastore Election " + leaderElection.getName());
- daemon.setDaemon(true);
- daemon.start();
- } else {
- daemon.run();
- }
+ daemon.setName("Metastore Election " + leaderElection.getName());
+ daemon.setDaemon(true);
+
daemon.setUncaughtExceptionHandler(newAbortOnElectionFailureHandler(leaderElection));
+ daemon.start();
+ daemons.add(daemon);
+ }
+ // Wait for each initial tryBeLeader() to complete so listeners have fired
before
+ // start() returns. Daemons are started before joining so elections still
run in parallel.
+ for (Thread daemon : daemons) {
+ daemon.join();
}
}
public void close() {
- leaderElections.forEach(le -> {
+ // Iterate over a snapshot: closing an election may fire listener
callbacks that
+ // touch the context, which could mutate leaderElections mid-iteration.
+ List<LeaderElection<?>> snapshot = new ArrayList<>(leaderElections);
+ snapshot.forEach(le -> {
try {
le.close();
} catch (Exception e) {
@@ -138,9 +150,53 @@ public void close() {
});
}
+ /**
+ * Builds the {@link Thread.UncaughtExceptionHandler} attached to each
election daemon.
+ * When the daemon terminates because {@code tryBeLeader} exhausted its
retries, this
+ * handler logs the failure and invokes {@link #abortAction}. The {@link
AtomicBoolean}
+ * guard ensures the abort runs at most once even if both electors fail
concurrently.
+ */
+ private Thread.UncaughtExceptionHandler newAbortOnElectionFailureHandler(
+ LeaderElection<?> leaderElection) {
+ return (t, ex) -> {
+ HiveMetaStore.LOG.error(
+ "Metastore leader election '{}' failed after {} retries in thread
'{}'; "
+ + "aborting Metastore to avoid running with unelected tasks.",
+ leaderElection.getName(),
+ MetastoreConf.getIntVar(conf,
MetastoreConf.ConfVars.LOCK_NUMRETRIES),
+ t.getName(),
+ ex);
+ if (isAborting.compareAndSet(false, true)) {
+ // Best-effort close the electors so the mutex/lease is released
promptly,
+ // letting other HMS instances proceed with election rather than
waiting for
+ // the JVM shutdown hook to release resources.
+ try {
+ close();
+ } catch (Exception e) {
+ HiveMetaStore.LOG.warn("Error closing elections during abort", e);
+ }
+ try {
+ abortAction.run();
+ } catch (Exception e) {
+ HiveMetaStore.LOG.error("Error while executing Metastore abort
action", e);
+ }
+ }
+ };
+ }
+
+ /**
+ * Overrides the action taken when a leader election exhausts its retries.
+ * Production code should leave the default ({@code System.exit(1)}), which
+ * triggers the HMS orderly-shutdown hook chain. Tests can inject a fake to
+ * assert the abort path fires without killing the test JVM.
+ */
+ @VisibleForTesting
+ void setAbortAction(Runnable abortAction) {
+ this.abortAction = requireNonNull(abortAction, "abortAction is null");
+ }
+
public static class ContextBuilder {
private Configuration configuration;
- private boolean startAsDaemon;
private String servHost;
private IHMSHandler handler;
private TTYPE ttype = TTYPE.HOUSEKEEPING;
@@ -183,14 +239,9 @@ public ContextBuilder addListener(
return this;
}
- public ContextBuilder startAsDaemon(boolean daemon) {
- this.startAsDaemon = daemon;
- return this;
- }
-
public LeaderElectionContext build() throws Exception {
return new LeaderElectionContext(servHost, configuration,
- listeners, startAsDaemon, handler);
+ listeners, handler);
}
}
}
diff --git
a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/leader/TestLeaderElection.java
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/leader/TestLeaderElection.java
index c3d24cd2dce..8d7b8dd28bc 100644
---
a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/leader/TestLeaderElection.java
+++
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/leader/TestLeaderElection.java
@@ -20,16 +20,23 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.common.TableName;
+import org.apache.hadoop.hive.metastore.annotation.MetastoreUnitTest;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.metastore.utils.TestTxnDbUtil;
import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
+@Category(MetastoreUnitTest.class)
public class TestLeaderElection {
@Test
@@ -107,4 +114,147 @@ public void testLeaseLeaderElection() throws Exception {
assertTrue(flag2.get() && instance2.isLeader());
}
+ /**
+ * HIVE-29720: when a leader election exhausts its retries the daemon thread
+ * previously died silently, leaving HMS running without housekeeping tasks.
+ * LeaderElectionContext now installs an UncaughtExceptionHandler that
invokes
+ * an abort action (System.exit(1) in production). This test injects a fake
+ * abort action to assert the handler fires when an election fails.
+ */
+ @Test
+ public void testAbortActionInvokedWhenElectionFails() throws Exception {
+ AtomicInteger invocations = new AtomicInteger(0);
+ LeaderElectionFactory.ElectionCreator originalHostCreator =
+ conf -> new StaticLeaderElection();
+ LeaderElectionFactory.addElectionCreator(
+ LeaderElectionFactory.Method.HOST, conf -> new
FailingLeaderElection(invocations));
+ try {
+ Configuration conf = newAbortTestConf();
+ CountDownLatch abortLatch = new CountDownLatch(1);
+
+ LeaderElectionContext context = new
LeaderElectionContext.ContextBuilder(conf)
+ .servHost("test-host")
+ .setTType(LeaderElectionContext.TTYPE.HOUSEKEEPING)
+ .addListener(new TestLeaderListener(new AtomicBoolean(false)))
+ .build();
+ context.setAbortAction(abortLatch::countDown);
+
+ context.start();
+
+ assertTrue("HMS should trigger abort action when leader election fails",
+ abortLatch.await(10, TimeUnit.SECONDS));
+ assertEquals("Failing election should have been invoked once",
+ 1, invocations.get());
+ } finally {
+ LeaderElectionFactory.addElectionCreator(
+ LeaderElectionFactory.Method.HOST, originalHostCreator);
+ }
+ }
+
+ /**
+ * HIVE-29720: when both electors (HOUSEKEEPING and ALWAYS_TASKS) fail near
+ * simultaneously, the abort action must still run at most once. Guarded by
+ * an AtomicBoolean in LeaderElectionContext.
+ */
+ @Test
+ public void testAbortActionInvokedOnceWhenBothElectorsFail() throws
Exception {
+ AtomicInteger invocations = new AtomicInteger(0);
+ LeaderElectionFactory.ElectionCreator originalHostCreator =
+ conf -> new StaticLeaderElection();
+ LeaderElectionFactory.addElectionCreator(
+ LeaderElectionFactory.Method.HOST, conf -> new
FailingLeaderElection(invocations));
+ try {
+ Configuration conf = newAbortTestConf();
+ AtomicInteger abortCount = new AtomicInteger(0);
+ CountDownLatch firstAbortLatch = new CountDownLatch(1);
+
+ LeaderElectionContext context = new
LeaderElectionContext.ContextBuilder(conf)
+ .servHost("test-host")
+ .setTType(LeaderElectionContext.TTYPE.HOUSEKEEPING)
+ .addListener(new TestLeaderListener(new AtomicBoolean(false)))
+ .setTType(LeaderElectionContext.TTYPE.ALWAYS_TASKS)
+ .addListener(new TestLeaderListener(new AtomicBoolean(false)))
+ .build();
+ context.setAbortAction(() -> {
+ abortCount.incrementAndGet();
+ firstAbortLatch.countDown();
+ });
+
+ context.start();
+
+ assertTrue("HMS should trigger abort action when either election fails",
+ firstAbortLatch.await(10, TimeUnit.SECONDS));
+
+ // Wait briefly so any race between the two failing daemons resolves,
+ // then verify the AtomicBoolean guard kept abortAction at a single
invocation.
+ Thread.sleep(500);
+
+ assertEquals("Abort action must run exactly once even if both electors
fail",
+ 1, abortCount.get());
+ assertEquals("Both failing electors should have been invoked",
+ 2, invocations.get());
+ } finally {
+ LeaderElectionFactory.addElectionCreator(
+ LeaderElectionFactory.Method.HOST, originalHostCreator);
+ }
+ }
+
+ private static Configuration newAbortTestConf() {
+ Configuration conf = MetastoreConf.newMetastoreConf();
+ // Use the HOST election method so we don't need a real database mutex.
+ MetastoreConf.setVar(conf,
+ MetastoreConf.ConfVars.METASTORE_HOUSEKEEPING_LEADER_ELECTION, "HOST");
+ // Clear the audit table so LeaderElectionContext doesn't try to build an
+ // AuditLeaderListener against a live handler in this unit test.
+ MetastoreConf.setVar(conf,
+ MetastoreConf.ConfVars.METASTORE_HOUSEKEEPING_LEADER_AUDITTABLE, "");
+ return conf;
+ }
+
+ /**
+ * A {@link LeaderElection} that unconditionally throws {@link
LeaderException}
+ * from {@code tryBeLeader}, simulating retry exhaustion in the real
+ * {@code LeaseLeaderElection}. Registered via
+ * {@link LeaderElectionFactory#addElectionCreator} for the duration of a
test.
+ */
+ static final class FailingLeaderElection implements LeaderElection<Object> {
+ private final AtomicInteger invocations;
+ private String name;
+
+ FailingLeaderElection(AtomicInteger invocations) {
+ this.invocations = invocations;
+ }
+
+ @Override
+ public void tryBeLeader(Configuration conf, Object mutex) throws
LeaderException {
+ invocations.incrementAndGet();
+ throw new LeaderException("Simulated retry exhaustion for election: " +
name);
+ }
+
+ @Override
+ public boolean isLeader() {
+ return false;
+ }
+
+ @Override
+ public void addStateListener(LeaderElection.LeadershipStateListener
listener) {
+ // no-op
+ }
+
+ @Override
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public void close() throws IOException {
+ // no-op
+ }
+ }
+
}
