This is an automated email from the ASF dual-hosted git repository.
gianm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 7b39b78b8ef fix: Release leadership on shutdown with k8s leader
election. (#20800)
7b39b78b8ef is described below
commit 7b39b78b8efd779b62b4f5bd2cde5a22f3a40903
Author: Gian Merlino <[email protected]>
AuthorDate: Fri Oct 30 10:11:43 2026 -0700
fix: Release leadership on shutdown with k8s leader election. (#20800)
We were missing a call to leaderElector.close() on service stop, which
would lead to a longer-than-necessary delay waiting for the new leader
to take over.
---
.../discovery/DefaultK8sLeaderElectorFactory.java | 6 ++
.../k8s/discovery/K8sLeaderElectorFactory.java | 5 ++
.../k8s/discovery/LeaderElectorAsyncWrapper.java | 10 ++-
.../k8s/discovery/K8sDruidLeaderSelectorTest.java | 82 ++++++++++++++++++++++
.../druid/indexing/overlord/DruidOverlord.java | 32 +++++++--
.../druid/indexing/overlord/http/OverlordTest.java | 37 +++++++++-
.../druid/server/coordinator/DruidCoordinator.java | 10 ++-
.../server/coordinator/DruidCoordinatorTest.java | 16 ++++-
8 files changed, 187 insertions(+), 11 deletions(-)
diff --git
a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sLeaderElectorFactory.java
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sLeaderElectorFactory.java
index 7ac6b5e54ba..47b4efebd42 100644
---
a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sLeaderElectorFactory.java
+++
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sLeaderElectorFactory.java
@@ -76,6 +76,12 @@ public class DefaultK8sLeaderElectorFactory implements
K8sLeaderElectorFactory
{
leaderElector.run(startLeadingHook, stopLeadingHook);
}
+
+ @Override
+ public void close()
+ {
+ leaderElector.close();
+ }
};
}
diff --git
a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sLeaderElectorFactory.java
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sLeaderElectorFactory.java
index 10bab39f65c..e7fd307145a 100644
---
a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sLeaderElectorFactory.java
+++
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sLeaderElectorFactory.java
@@ -38,4 +38,9 @@ interface K8sLeaderElector
{
String getCurrentLeader();
void run(Runnable startLeadingHook, Runnable stopLeadingHook);
+
+ /**
+ * Stops this elector permanently and, if it holds the lock, releases it.
+ */
+ void close();
}
diff --git
a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/LeaderElectorAsyncWrapper.java
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/LeaderElectorAsyncWrapper.java
index f7e1b481d04..7a791c93124 100644
---
a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/LeaderElectorAsyncWrapper.java
+++
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/LeaderElectorAsyncWrapper.java
@@ -103,8 +103,14 @@ public class LeaderElectorAsyncWrapper implements Closeable
try {
futureRef.get().cancel(true);
executor.shutdownNow();
- if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
- LOGGER.warn("Failed to terminate [%s] executor.",
this.getClass().getSimpleName());
+ if (executor.awaitTermination(10, TimeUnit.SECONDS)) {
+ // Release the lock after the run loop has exited.
+ k8sLeaderElector.close();
+ } else {
+ LOGGER.warn(
+ "Failed to terminate [%s] executor. Not releasing the lock; it
will expire after the lease duration.",
+ this.getClass().getSimpleName()
+ );
}
}
catch (InterruptedException ex) {
diff --git
a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidLeaderSelectorTest.java
b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidLeaderSelectorTest.java
index 8482117129e..d78f76ebd9e 100644
---
a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidLeaderSelectorTest.java
+++
b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidLeaderSelectorTest.java
@@ -28,6 +28,8 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -77,6 +79,11 @@ public class K8sDruidLeaderSelectorTest
stopLeadingHook.run();
}
}
+
+ @Override
+ public void close()
+ {
+ }
};
}
}
@@ -150,6 +157,11 @@ public class K8sDruidLeaderSelectorTest
}
}
}
+
+ @Override
+ public void close()
+ {
+ }
};
}
}
@@ -181,4 +193,74 @@ public class K8sDruidLeaderSelectorTest
leaderSelector.unregisterListener();
stopBeingLeaderLatch.await();
}
+
+ @Test
+ @Timeout(value = 5_000, unit = TimeUnit.MILLISECONDS)
+ public void
testLeaderElection_unregisterReleasesLockAfterStoppingLeadership() throws
Exception
+ {
+ final List<String> events = new CopyOnWriteArrayList<>();
+
+ K8sDruidLeaderSelector leaderSelector = new K8sDruidLeaderSelector(
+ testNode1.getDruidNode(),
+ lockResourceName,
+ discoveryConfig.getCoordinatorLeaderElectionConfigMapNamespace(),
+ discoveryConfig,
+ new K8sLeaderElectorFactory()
+ {
+ @Override
+ public K8sLeaderElector create(String candidateId, String namespace,
String lockResourceName)
+ {
+ return new K8sLeaderElector()
+ {
+ @Override
+ public String getCurrentLeader()
+ {
+ return testNode1.getDruidNode().getHostAndPortToUse();
+ }
+
+ @Override
+ public void run(Runnable startLeadingHook, Runnable
stopLeadingHook)
+ {
+ startLeadingHook.run();
+ try {
+ Thread.sleep(Long.MAX_VALUE);
+ }
+ catch (InterruptedException ex) {
+ stopLeadingHook.run();
+ }
+ }
+
+ @Override
+ public void close()
+ {
+ events.add("close");
+ }
+ };
+ }
+ }
+ );
+
+ CountDownLatch becomeLeaderLatch = new CountDownLatch(1);
+
+ leaderSelector.registerListener(
+ new DruidLeaderSelector.Listener()
+ {
+ @Override
+ public void becomeLeader()
+ {
+ becomeLeaderLatch.countDown();
+ }
+
+ @Override
+ public void stopBeingLeader()
+ {
+ events.add("stopBeingLeader");
+ }
+ }
+ );
+
+ becomeLeaderLatch.await();
+ leaderSelector.unregisterListener();
+ Assertions.assertEquals(List.of("stopBeingLeader", "close"), events);
+ }
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/DruidOverlord.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/DruidOverlord.java
index 6e7d25168a7..bcded57e0ff 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/DruidOverlord.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/DruidOverlord.java
@@ -21,6 +21,7 @@ package org.apache.druid.indexing.overlord;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Optional;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
import com.google.inject.Inject;
import org.apache.druid.client.indexing.IndexingService;
import org.apache.druid.discovery.DruidLeaderSelector;
@@ -69,6 +70,12 @@ public class DruidOverlord
*/
private volatile boolean initialized;
+ /**
+ * Whether {@link #stop()} has been called.
+ */
+ @GuardedBy("giant")
+ private boolean stopped;
+
@Inject
public DruidOverlord(
final TaskMaster taskMaster,
@@ -102,10 +109,15 @@ public class DruidOverlord
{
giant.lock();
- // I AM THE MASTER OF THE UNIVERSE.
- log.info("By the power of Grayskull, I have the power. I am the
leader");
-
try {
+ if (stopped) {
+ log.info("Not becoming leader, since this Overlord is stopped.");
+ return;
+ }
+
+ // I AM THE MASTER OF THE UNIVERSE.
+ log.info("By the power of Grayskull, I have the power. I am the
leader");
+
final TaskRunner taskRunner = runnerFactory.build();
final TaskQueue taskQueue = new TaskQueue(
taskLockConfig,
@@ -234,8 +246,20 @@ public class DruidOverlord
giant.lock();
try {
+ stopped = true;
gracefulStopLeaderLifecycle();
- overlordLeaderSelector.unregisterListener();
+ }
+ finally {
+ giant.unlock();
+ }
+
+ // Unregister without holding the giant lock. When using
K8sDruidLeaderSelector, it will call
+ // stopBeingLeader() from another thread and wait for it, which needs the
same lock.
+ overlordLeaderSelector.unregisterListener();
+
+ giant.lock();
+
+ try {
if (isStandalone()) {
segmentMetadataCache.stop();
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java
index 822362cf59e..1d9805b4b13 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java
@@ -113,6 +113,7 @@ public class OverlordTest
private Timing timing;
private CuratorFramework curator;
private DruidOverlord overlord;
+ private TestDruidLeaderSelector leaderSelector;
private TaskMaster taskMaster;
private GlobalTaskLockbox taskLockbox;
private TaskStorage taskStorage;
@@ -243,6 +244,7 @@ public class OverlordTest
taskActionClientFactory,
supervisorManager
);
+ leaderSelector = new TestDruidLeaderSelector();
overlord = new DruidOverlord(
taskMaster,
new TaskLockConfig(),
@@ -256,7 +258,7 @@ public class OverlordTest
serviceEmitter,
supervisorManager,
EasyMock.createNiceMock(OverlordDutyExecutor.class),
- new TestDruidLeaderSelector(),
+ leaderSelector,
EasyMock.createNiceMock(SegmentAllocationQueue.class),
EasyMock.createNiceMock(SegmentMetadataCache.class),
Set.of(
@@ -373,6 +375,23 @@ public class OverlordTest
EasyMock.verify(taskActionClientFactory);
}
+ @Timeout(60)
+ @Test
+ public void testBecomeLeaderAfterStopIsIgnored() throws Exception
+ {
+ overlord.start();
+ while (!overlord.isLeader()) {
+ Thread.sleep(10);
+ }
+
+ overlord.stop();
+ Assertions.assertFalse(taskMaster.isHalfOrFullLeader());
+
+ // Simulate a becomeLeader() callback that the leader selector delivers
after stop().
+ leaderSelector.listener.becomeLeader();
+ Assertions.assertFalse(taskMaster.isHalfOrFullLeader());
+ }
+
/* Wait until the task with given taskId has the given Task Status
* These method will not timeout until the condition is met so calling
method should ensure timeout
* This method also assumes that the task with given taskId is present
@@ -602,11 +621,25 @@ public class OverlordTest
listener.becomeLeader();
}
+ /**
+ * Calls {@link Listener#stopBeingLeader()} from another thread and waits
for it, as K8sDruidLeaderSelector
+ * does, so callers that hold a lock needed by stopBeingLeader() fail here
instead of deadlocking.
+ */
@Override
public void unregisterListener()
{
leader = null;
- listener.stopBeingLeader();
+
+ final Thread thread = new Thread(listener::stopBeingLeader);
+ thread.start();
+ try {
+ thread.join(10_000);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ Assertions.assertFalse(thread.isAlive(), "stopBeingLeader() did not
finish");
}
}
}
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java
b/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java
index d385cad797a..ef4ab2244aa 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java
@@ -401,9 +401,15 @@ public class DruidCoordinator
return;
}
- coordLeaderSelector.unregisterListener();
- metadataManager.stopCache();
started = false;
+ }
+
+ // Unregister without holding the lock. When using K8sDruidLeaderSelector,
it will call
+ // stopBeingLeader() from another thread and wait for it, which needs the
same lock.
+ coordLeaderSelector.unregisterListener();
+
+ synchronized (lock) {
+ metadataManager.stopCache();
stopAllDutyGroups();
}
}
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
index c3203df0a93..a310957a68d 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
@@ -1127,11 +1127,25 @@ public class DruidCoordinatorTest
listener.becomeLeader();
}
+ /**
+ * Calls {@link Listener#stopBeingLeader()} from another thread and waits
for it, as K8sDruidLeaderSelector
+ * does, so callers that hold a lock needed by stopBeingLeader() fail here
instead of deadlocking.
+ */
@Override
public void unregisterListener()
{
leader = null;
- listener.stopBeingLeader();
+
+ final Thread thread = new Thread(listener::stopBeingLeader);
+ thread.start();
+ try {
+ thread.join(10_000);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ Assertions.assertFalse(thread.isAlive(), "stopBeingLeader() did not
finish");
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
