You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by ka...@apache.org on 2026/07/25 11:11:10 UTC
(druid) branch master updated: fix: Avoid sleep on stop in K8sDruidNodeDiscoveryProvider. (#19748)
This is an automated email from the ASF dual-hosted git repository.
cryptoe 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 67aa25cef0d fix: Avoid sleep on stop in K8sDruidNodeDiscoveryProvider. (#19748)
67aa25cef0d is described below
commit 67aa25cef0d0788b89062838871e2964e3a52285
Author: Gian Merlino <gi...@gmail.com>
AuthorDate: Sat Jul 25 04:11:04 2026 -0700
fix: Avoid sleep on stop in K8sDruidNodeDiscoveryProvider. (#19748)
Prior to this patch, an InterruptedException caused by shutdownNow()
in stop() would trigger sleep(watcherErrorRetryWaitMS), leading to a
ten second delay on stop().
---
.../discovery/K8sDruidNodeDiscoveryProvider.java | 18 +++--
.../K8sDruidNodeDiscoveryProviderTest.java | 94 ++++++++++++++++++++++
2 files changed, 106 insertions(+), 6 deletions(-)
diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java
index 96497d4510f..8dfc595db74 100644
--- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java
+++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java
@@ -223,7 +223,7 @@ public class K8sDruidNodeDiscoveryProvider extends DruidNodeDiscoveryProvider
return;
}
- while (lifecycleLock.awaitStarted(1, TimeUnit.MILLISECONDS)) {
+ while (lifecycleLock.isStarted()) {
try {
DiscoveryDruidNodeList list = k8sApiClient.listPods(podInfo.getPodNamespace(), labelSelector, nodeRole);
baseNodeRoleWatcher.resetNodes(list.getDruidNodes());
@@ -241,8 +241,10 @@ public class K8sDruidNodeDiscoveryProvider extends DruidNodeDiscoveryProvider
catch (Throwable ex) {
LOGGER.error(ex, "Exception while watching for role[%s].", nodeRole);
- // Wait a little before trying again.
- sleep(watcherErrorRetryWaitMS);
+ if (lifecycleLock.isStarted()) {
+ // If not stopped, wait a little before trying again.
+ sleep(watcherErrorRetryWaitMS);
+ }
}
}
@@ -252,7 +254,7 @@ public class K8sDruidNodeDiscoveryProvider extends DruidNodeDiscoveryProvider
private void keepWatching(String labelSelector, String resourceVersion)
{
String nextResourceVersion = resourceVersion;
- while (lifecycleLock.awaitStarted(1, TimeUnit.MILLISECONDS)) {
+ while (lifecycleLock.isStarted()) {
try {
WatchResult iter =
k8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector, nextResourceVersion, nodeRole);
@@ -300,11 +302,15 @@ public class K8sDruidNodeDiscoveryProvider extends DruidNodeDiscoveryProvider
catch (SocketTimeoutException ex) {
// socket read timeout can happen normally due to k8s not having anything new to push leading to socket
// read timeout, so no error log
- sleep(watcherErrorRetryWaitMS);
+ if (lifecycleLock.isStarted()) {
+ sleep(watcherErrorRetryWaitMS);
+ }
}
catch (Throwable ex) {
LOGGER.error(ex, "Error while watching role[%s]", this.nodeRole);
- sleep(watcherErrorRetryWaitMS);
+ if (lifecycleLock.isStarted()) {
+ sleep(watcherErrorRetryWaitMS);
+ }
}
}
}
diff --git a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java
index 18e030f2507..6417edb4b27 100644
--- a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java
+++ b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java
@@ -40,6 +40,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
public class K8sDruidNodeDiscoveryProviderTest
@@ -430,6 +431,61 @@ public class K8sDruidNodeDiscoveryProviderTest
discoveryProvider.stop();
}
+ @Test
+ @Timeout(value = 60_000, unit = TimeUnit.MILLISECONDS)
+ public void testStopDoesNotWaitForErrorRetryWhenWatchIsInterrupted() throws Exception
+ {
+ final String labelSelector = "druidDiscoveryAnnouncement-cluster-identifier=druid-cluster,druidDiscoveryAnnouncement-router=true";
+
+ // Long enough that a single unguarded retry sleep outlasts the 15s stop() timeout on watchExecutor.
+ final long watcherErrorRetryWaitMS = 30_000;
+
+ final CountDownLatch watchBlocked = new CountDownLatch(1);
+
+ final K8sApiClient mockK8sApiClient = EasyMock.createMock(K8sApiClient.class);
+ EasyMock.expect(mockK8sApiClient.listPods(podInfo.getPodNamespace(), labelSelector, NodeRole.ROUTER)).andReturn(
+ new DiscoveryDruidNodeList(
+ "v1",
+ ImmutableMap.of(testNode1.getDruidNode().getHostAndPortToUse(), testNode1)
+ )
+ );
+ EasyMock.expect(mockK8sApiClient.watchPods(
+ podInfo.getPodNamespace(), labelSelector, "v1", NodeRole.ROUTER)).andReturn(
+ new BlockingWatchResult(watchBlocked)
+ );
+ EasyMock.replay(mockK8sApiClient);
+
+ final K8sDruidNodeDiscoveryProvider discoveryProvider = new K8sDruidNodeDiscoveryProvider(
+ podInfo,
+ discoveryConfig,
+ mockK8sApiClient,
+ watcherErrorRetryWaitMS
+ );
+ discoveryProvider.start();
+
+ final K8sDruidNodeDiscoveryProvider.NodeRoleWatcher nodeDiscovery =
+ discoveryProvider.getForNodeRole(NodeRole.ROUTER, false);
+ nodeDiscovery.start();
+
+ Assertions.assertTrue(
+ watchBlocked.await(30, TimeUnit.SECONDS),
+ "watch thread never reached hasNext()"
+ );
+
+ final long startNs = System.nanoTime();
+ discoveryProvider.stop();
+ final long stopMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
+
+ Assertions.assertTrue(
+ stopMillis < 5_000,
+ StringUtils.format(
+ "stop() took [%,d] ms, expected it not to wait out watcherErrorRetryWaitMS[%,d]",
+ stopMillis,
+ watcherErrorRetryWaitMS
+ )
+ );
+ }
+
private static class MockListener implements DruidNodeDiscovery.Listener
{
List<Event> events;
@@ -554,6 +610,44 @@ public class K8sDruidNodeDiscoveryProviderTest
}
}
+ /**
+ * A {@link WatchResult} that parks in {@link #hasNext()} the way an open k8s watch socket does, and surfaces the
+ * shutdown interrupt as a {@link RuntimeException}.
+ */
+ private static class BlockingWatchResult implements WatchResult
+ {
+ private final CountDownLatch blocked;
+
+ public BlockingWatchResult(CountDownLatch blocked)
+ {
+ this.blocked = blocked;
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ blocked.countDown();
+ try {
+ Thread.sleep(Long.MAX_VALUE);
+ return false; // Not reached.
+ }
+ catch (InterruptedException ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ @Override
+ public Watch.Response<DiscoveryDruidNodeAndResourceVersion> next()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close()
+ {
+ }
+ }
+
private static class MockWatchResult implements WatchResult
{
private List<Watch.Response<DiscoveryDruidNodeAndResourceVersion>> results;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org