From b31a68181293d3af43ad9c07c7db89b42397d096 Mon Sep 17 00:00:00 2001 From: Vivek Dhiman Date: Thu, 26 Feb 2026 04:22:29 +0530 Subject: [PATCH 1/7] Restarted the watch after stream reset --- .../K8sDruidNodeDiscoveryProvider.java | 64 +++++++++---------- .../druid/k8s/discovery/WatchResult.java | 4 +- 2 files changed, 30 insertions(+), 38 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 8dfc595db742..db376c79eff2 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 @@ -23,6 +23,7 @@ import com.google.common.base.Preconditions; import com.google.inject.Inject; import io.kubernetes.client.util.Watch; +import okhttp3.internal.http2.StreamResetException; import org.apache.druid.concurrent.LifecycleLock; import org.apache.druid.discovery.BaseNodeRoleWatcher; import org.apache.druid.discovery.DiscoveryDruidNode; @@ -254,50 +255,43 @@ private void watch() private void keepWatching(String labelSelector, String resourceVersion) { String nextResourceVersion = resourceVersion; - while (lifecycleLock.isStarted()) { - try { - WatchResult iter = - k8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector, nextResourceVersion, nodeRole); + while (lifecycleLock.awaitStarted(1, TimeUnit.MILLISECONDS)) { + + try (WatchResult iter = k8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector, nextResourceVersion, nodeRole)) { if (iter == null) { // history not available, we need to start from scratch return; } - try { - while (iter.hasNext()) { - Watch.Response item = iter.next(); - if (item != null && item.type != null && item.object != null) { - switch (item.type) { - case WatchResult.ADDED: - baseNodeRoleWatcher.childAdded(item.object.getNode()); - break; - case WatchResult.DELETED: - case WatchResult.NOT_READY: - // Use skipIfUnknown=true for all k8s discovery removals. - // DELETED can fire after NOT_READY (so the service is already removed), or before ADDED (pod deleted before becoming ready). - // NOT_READY can repeat during CrashLoopBackOff. None of these warrant the error-level logging that - // comes with trying to remove an unknown service. - baseNodeRoleWatcher.childRemoved(item.object.getNode(), true); - break; - default: - } - - // This should be updated after the action has been dealt with successfully - nextResourceVersion = item.object.getResourceVersion(); - - } else { - // Try again by starting the watch from the beginning. This can happen if the - // watch goes bad. - LOGGER.debug("Received NULL item while watching role[%s]. Restarting watch.", this.nodeRole); - return; - } + while (iter.hasNext()) { + Watch.Response item = iter.next(); + + if (item == null || item.type == null || item.object == null) { + LOGGER.debug("Received NULL item while watching role[%s]. Restarting watch.", this.nodeRole); + return; } - } - finally { - iter.close(); + + switch (item.type) { + case WatchResult.ADDED: + baseNodeRoleWatcher.childAdded(item.object.getNode()); + break; + case WatchResult.DELETED: + baseNodeRoleWatcher.childRemoved(item.object.getNode()); + break; + default: + } + + // This should be updated after the action has been dealt with successfully + nextResourceVersion = item.object.getResourceVersion(); } + LOGGER.trace("Watch closed normally for role[%s]", this.nodeRole); + return; + } + catch (StreamResetException ex) { + LOGGER.debug("Watch stream terminated normally for role[%s], restarting", this.nodeRole); + return; } catch (SocketTimeoutException ex) { // socket read timeout can happen normally due to k8s not having anything new to push leading to socket diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java index 2b3ef7b584a9..e90bbcfb08c4 100644 --- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java +++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java @@ -29,7 +29,7 @@ * a Kubernetes MODIFIED event for a pod whose containers are no longer ready (according to k8s readiness state) is surfaced as the * synthetic {@link #NOT_READY} type so that consumers can handle it as a removal from k8s service discovery. */ -public interface WatchResult +public interface WatchResult extends AutoCloseable { String ADDED = "ADDED"; String MODIFIED = "MODIFIED"; @@ -47,6 +47,4 @@ public interface WatchResult boolean hasNext() throws SocketTimeoutException; Watch.Response next(); - - void close(); } From 3df0b36d516c494b98abc869c11f00dd1e23b167 Mon Sep 17 00:00:00 2001 From: Vivek Dhiman Date: Wed, 4 Mar 2026 15:55:26 +0530 Subject: [PATCH 2/7] Addressed review comments --- .../druid/k8s/discovery/ChannelResetException.java | 10 ++++++++++ .../druid/k8s/discovery/DefaultK8sApiClient.java | 9 +++++++-- .../k8s/discovery/K8sDruidNodeDiscoveryProvider.java | 3 +-- .../org/apache/druid/k8s/discovery/WatchResult.java | 4 ++-- 4 files changed, 20 insertions(+), 6 deletions(-) create mode 100644 extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/ChannelResetException.java diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/ChannelResetException.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/ChannelResetException.java new file mode 100644 index 000000000000..38075f85a1ad --- /dev/null +++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/ChannelResetException.java @@ -0,0 +1,10 @@ +package org.apache.druid.k8s.discovery; + +import java.io.IOException; + +public class ChannelResetException extends IOException +{ + public ChannelResetException(Throwable ex) { + super(ex); + } +} diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java index b7992a3fb2ad..e6789493f7e5 100644 --- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java +++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java @@ -33,6 +33,7 @@ import io.kubernetes.client.openapi.models.V1PodList; import io.kubernetes.client.util.PatchUtils; import io.kubernetes.client.util.Watch; +import okhttp3.internal.http2.StreamResetException; import org.apache.druid.discovery.DiscoveryDruidNode; import org.apache.druid.discovery.NodeRole; import org.apache.druid.guice.annotations.Json; @@ -194,7 +195,7 @@ public WatchResult watchPods(String namespace, String labelSelector, String last private Watch.Response obj; @Override - public boolean hasNext() throws SocketTimeoutException + public boolean hasNext() throws IOException { try { while (watch.hasNext()) { @@ -274,7 +275,11 @@ public boolean hasNext() throws SocketTimeoutException catch (RuntimeException ex) { if (ex.getCause() instanceof SocketTimeoutException) { throw (SocketTimeoutException) ex.getCause(); - } else { + } + if (ex.getCause() instanceof StreamResetException) { + throw new ChannelResetException(ex.getCause()); + } + else { throw ex; } } 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 db376c79eff2..42ac9a34026f 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 @@ -23,7 +23,6 @@ import com.google.common.base.Preconditions; import com.google.inject.Inject; import io.kubernetes.client.util.Watch; -import okhttp3.internal.http2.StreamResetException; import org.apache.druid.concurrent.LifecycleLock; import org.apache.druid.discovery.BaseNodeRoleWatcher; import org.apache.druid.discovery.DiscoveryDruidNode; @@ -289,7 +288,7 @@ private void keepWatching(String labelSelector, String resourceVersion) LOGGER.trace("Watch closed normally for role[%s]", this.nodeRole); return; } - catch (StreamResetException ex) { + catch (ChannelResetException ex) { LOGGER.debug("Watch stream terminated normally for role[%s], restarting", this.nodeRole); return; } diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java index e90bbcfb08c4..37cac3768c56 100644 --- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java +++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java @@ -21,7 +21,7 @@ import io.kubernetes.client.util.Watch; -import java.net.SocketTimeoutException; +import java.io.IOException; /** * Iterator over k8s pod watch events that is aligned with the needs of Druid service discovery rather than @@ -44,7 +44,7 @@ public interface WatchResult extends AutoCloseable */ String NOT_READY = "NOT_READY"; - boolean hasNext() throws SocketTimeoutException; + boolean hasNext() throws IOException; Watch.Response next(); } From bc2f2d98d6220a437e855b9f4a237480c3066a8e Mon Sep 17 00:00:00 2001 From: Vivek Dhiman Date: Wed, 4 Mar 2026 16:25:06 +0530 Subject: [PATCH 3/7] Add unit test and addresses styling issues --- extensions-core/kubernetes-extensions/pom.xml | 5 ++ .../k8s/discovery/ChannelResetException.java | 22 ++++- .../k8s/discovery/DefaultK8sApiClient.java | 4 +- .../K8sDruidNodeDiscoveryProviderTest.java | 84 ++++++++++++++++++- 4 files changed, 110 insertions(+), 5 deletions(-) diff --git a/extensions-core/kubernetes-extensions/pom.xml b/extensions-core/kubernetes-extensions/pom.xml index 7decd1a881f7..b9e8a310ff92 100644 --- a/extensions-core/kubernetes-extensions/pom.xml +++ b/extensions-core/kubernetes-extensions/pom.xml @@ -100,6 +100,11 @@ okhttp-jvm + + com.squareup.okhttp3 + okhttp + + org.junit.jupiter diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/ChannelResetException.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/ChannelResetException.java index 38075f85a1ad..345ec04fed92 100644 --- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/ChannelResetException.java +++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/ChannelResetException.java @@ -1,10 +1,30 @@ +/* + * 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.druid.k8s.discovery; import java.io.IOException; public class ChannelResetException extends IOException { - public ChannelResetException(Throwable ex) { + public ChannelResetException(Throwable ex) + { super(ex); } } diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java index e6789493f7e5..af0df6f26c7a 100644 --- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java +++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java @@ -279,9 +279,7 @@ public boolean hasNext() throws IOException if (ex.getCause() instanceof StreamResetException) { throw new ChannelResetException(ex.getCause()); } - else { - throw ex; - } + throw ex; } return false; 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 6417edb4b275..88b7a377e922 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 @@ -34,6 +34,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; +import java.io.IOException; import java.net.SocketTimeoutException; import java.util.ArrayList; import java.util.Collection; @@ -486,6 +487,71 @@ public void testStopDoesNotWaitForErrorRetryWhenWatchIsInterrupted() throws Exce ); } + @Test + @Timeout(value = 10_000, unit = TimeUnit.MILLISECONDS) + public void testNodeRoleWatcherHandlesChannelResetException() throws Exception + { + String labelSelector = "druidDiscoveryAnnouncement-cluster-identifier=druid-cluster,druidDiscoveryAnnouncement-router=true"; + 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, + testNode2.getDruidNode().getHostAndPortToUse(), testNode2 + ) + ) + ); + + EasyMock.expect(mockK8sApiClient.watchPods( + podInfo.getPodNamespace(), labelSelector, "v1", NodeRole.ROUTER)).andReturn( + new MockWatchResult(Collections.emptyList(), false, false, true) + ); + + EasyMock.expect(mockK8sApiClient.listPods(podInfo.getPodNamespace(), labelSelector, NodeRole.ROUTER)).andReturn( + new DiscoveryDruidNodeList( + "v2", + ImmutableMap.of( + testNode2.getDruidNode().getHostAndPortToUse(), testNode2, + testNode3.getDruidNode().getHostAndPortToUse(), testNode3 + ) + ) + ); + + EasyMock.expect(mockK8sApiClient.watchPods( + podInfo.getPodNamespace(), labelSelector, "v2", NodeRole.ROUTER)).andReturn( + new MockWatchResult(Collections.emptyList(), false, false) + ); + + EasyMock.replay(mockK8sApiClient); + + K8sDruidNodeDiscoveryProvider discoveryProvider = new K8sDruidNodeDiscoveryProvider( + podInfo, + discoveryConfig, + mockK8sApiClient, + 1 + ); + discoveryProvider.start(); + + K8sDruidNodeDiscoveryProvider.NodeRoleWatcher nodeDiscovery = + discoveryProvider.getForNodeRole(NodeRole.ROUTER, false); + + MockListener testListener = new MockListener( + ImmutableList.of( + MockListener.Event.added(testNode1), + MockListener.Event.added(testNode2), + MockListener.Event.inited(), + MockListener.Event.added(testNode3), + MockListener.Event.deleted(testNode1) + ) + ); + nodeDiscovery.registerListener(testListener); + nodeDiscovery.start(); + testListener.assertSuccess(); + discoveryProvider.stop(); + } + private static class MockListener implements DruidNodeDiscovery.Listener { List events; @@ -654,6 +720,7 @@ private static class MockWatchResult implements WatchResult private volatile boolean timeoutOnStart; private volatile boolean timeooutOnEmptyResults; + private volatile boolean channelResetOnStart; private volatile boolean closeCalled = false; public MockWatchResult( @@ -667,13 +734,28 @@ public MockWatchResult( this.timeooutOnEmptyResults = timeooutOnEmptyResults; } + public MockWatchResult( + List> results, + boolean timeoutOnStart, + boolean timeooutOnEmptyResults, + boolean channelResetOnStart + ) + { + this(results, timeoutOnStart, timeooutOnEmptyResults); + this.channelResetOnStart = channelResetOnStart; + } + @Override - public boolean hasNext() throws SocketTimeoutException + public boolean hasNext() throws IOException { if (timeoutOnStart) { throw new SocketTimeoutException("testing timeout on start!!!"); } + if (channelResetOnStart) { // ADD THIS BLOCK + throw new ChannelResetException(new RuntimeException("simulated StreamResetException")); + } + if (results.isEmpty()) { if (timeooutOnEmptyResults) { throw new SocketTimeoutException("testing timeout on end!!!"); From 863560957553af76353b0e5389c223e6f9a6a501 Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Wed, 18 Mar 2026 16:57:21 -0500 Subject: [PATCH 4/7] Fix service discovery bug in `kubernetes-extensions` (#19139) --- .../K8sDruidNodeDiscoveryProvider.java | 7 +- .../discovery/BaseNodeRoleWatcherTest.java | 66 +++++++++++++++++++ 2 files changed, 72 insertions(+), 1 deletion(-) 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 42ac9a34026f..bccf5eae2c0e 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 @@ -276,7 +276,12 @@ private void keepWatching(String labelSelector, String resourceVersion) baseNodeRoleWatcher.childAdded(item.object.getNode()); break; case WatchResult.DELETED: - baseNodeRoleWatcher.childRemoved(item.object.getNode()); + case WatchResult.NOT_READY: + // Use skipIfUnknown=true for all k8s discovery removals. + // DELETED can fire after NOT_READY (so the service is already removed), or before ADDED (pod deleted before becoming ready). + // NOT_READY can repeat during CrashLoopBackOff. None of these warrant the error-level logging that + // comes with trying to remove an unknown service. + baseNodeRoleWatcher.childRemoved(item.object.getNode(), true); break; default: } diff --git a/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java b/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java index 8295a3e9bd7f..d79237f12677 100644 --- a/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java +++ b/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java @@ -319,6 +319,72 @@ public void testChildRemovedIfPresentRepeatedRemovalsAreIdempotent() Assertions.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList); } + @Test(timeout = 60_000L) + public void testChildRemovedIfPresentRemovesKnownNode() + { + BaseNodeRoleWatcher nodeRoleWatcher = BaseNodeRoleWatcher.create(exec, NodeRole.BROKER); + + DiscoveryDruidNode broker1 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker1"); + + nodeRoleWatcher.childAdded(broker1); + nodeRoleWatcher.cacheInitialized(); + + TestListener listener = new TestListener(); + nodeRoleWatcher.registerListener(listener); + + Assert.assertEquals(ImmutableList.of(broker1), listener.nodesAddedList); + + // Remove with skipIfUnknown=true — node IS in cache, should remove and notify + nodeRoleWatcher.childRemoved(broker1, true); + + Assert.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList); + Assert.assertTrue(nodeRoleWatcher.getAllNodes().isEmpty()); + } + + @Test(timeout = 60_000L) + public void testChildRemovedIfPresentSkipsUnknownNode() + { + BaseNodeRoleWatcher nodeRoleWatcher = BaseNodeRoleWatcher.create(exec, NodeRole.BROKER); + + DiscoveryDruidNode broker1 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker1"); + DiscoveryDruidNode broker2 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker2"); + + nodeRoleWatcher.childAdded(broker1); + nodeRoleWatcher.cacheInitialized(); + + TestListener listener = new TestListener(); + nodeRoleWatcher.registerListener(listener); + + // Remove broker2 with skipIfUnknown=true — node is NOT in cache, should silently skip + nodeRoleWatcher.childRemoved(broker2, true); + + Assert.assertTrue(listener.nodesRemovedList.isEmpty()); + Assert.assertEquals(1, nodeRoleWatcher.getAllNodes().size()); + } + + @Test(timeout = 60_000L) + public void testChildRemovedIfPresentRepeatedRemovalsAreIdempotent() + { + BaseNodeRoleWatcher nodeRoleWatcher = BaseNodeRoleWatcher.create(exec, NodeRole.BROKER); + + DiscoveryDruidNode broker1 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker1"); + + nodeRoleWatcher.childAdded(broker1); + nodeRoleWatcher.cacheInitialized(); + + TestListener listener = new TestListener(); + nodeRoleWatcher.registerListener(listener); + + // First removal should remove and notify + nodeRoleWatcher.childRemoved(broker1, true); + Assert.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList); + + // Second removal should silently skip (node already removed) + nodeRoleWatcher.childRemoved(broker1, true); + // Still only one removal notification + Assert.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList); + } + private DiscoveryDruidNode buildDiscoveryDruidNode(NodeRole role, String host) { return new DiscoveryDruidNode( From 5b28bc3c76496e4d97facbbff6331f61776c1064 Mon Sep 17 00:00:00 2001 From: Vivek Dhiman Date: Tue, 28 Apr 2026 09:30:38 +0530 Subject: [PATCH 5/7] Address review comments. Remove hard dependency of okhttp3 internals --- extensions-core/kubernetes-extensions/pom.xml | 5 ----- .../org/apache/druid/k8s/discovery/DefaultK8sApiClient.java | 4 ++-- .../k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java | 2 +- 3 files changed, 3 insertions(+), 8 deletions(-) diff --git a/extensions-core/kubernetes-extensions/pom.xml b/extensions-core/kubernetes-extensions/pom.xml index b9e8a310ff92..7decd1a881f7 100644 --- a/extensions-core/kubernetes-extensions/pom.xml +++ b/extensions-core/kubernetes-extensions/pom.xml @@ -100,11 +100,6 @@ okhttp-jvm - - com.squareup.okhttp3 - okhttp - - org.junit.jupiter diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java index af0df6f26c7a..f9c30e2e4ad0 100644 --- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java +++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java @@ -33,7 +33,6 @@ import io.kubernetes.client.openapi.models.V1PodList; import io.kubernetes.client.util.PatchUtils; import io.kubernetes.client.util.Watch; -import okhttp3.internal.http2.StreamResetException; import org.apache.druid.discovery.DiscoveryDruidNode; import org.apache.druid.discovery.NodeRole; import org.apache.druid.guice.annotations.Json; @@ -41,6 +40,7 @@ import org.apache.druid.java.util.common.logger.Logger; import java.io.IOException; +import java.io.InterruptedIOException; import java.net.SocketTimeoutException; import java.util.HashMap; import java.util.List; @@ -276,7 +276,7 @@ public boolean hasNext() throws IOException if (ex.getCause() instanceof SocketTimeoutException) { throw (SocketTimeoutException) ex.getCause(); } - if (ex.getCause() instanceof StreamResetException) { + if (ex.getCause() instanceof IOException && !(ex.getCause() instanceof InterruptedIOException)) { throw new ChannelResetException(ex.getCause()); } throw ex; 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 88b7a377e922..a1a8b57ae0b7 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 @@ -752,7 +752,7 @@ public boolean hasNext() throws IOException throw new SocketTimeoutException("testing timeout on start!!!"); } - if (channelResetOnStart) { // ADD THIS BLOCK + if (channelResetOnStart) { throw new ChannelResetException(new RuntimeException("simulated StreamResetException")); } From cbcc766393f1341731869a24f59255ee833e7680 Mon Sep 17 00:00:00 2001 From: Vivek Dhiman Date: Tue, 1 Sep 2026 12:46:31 +0530 Subject: [PATCH 6/7] fix: remove duplicate JUnit4 tests reintroduced by rebase Rebasing onto apache/druid master silently duplicated three childRemoved(skipIfUnknown) tests: the JUnit5 versions already in this branch's history merged alongside stale JUnit4 copies from before the junit5-migration commit, instead of conflicting. --- .../discovery/BaseNodeRoleWatcherTest.java | 66 ------------------- 1 file changed, 66 deletions(-) diff --git a/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java b/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java index d79237f12677..8295a3e9bd7f 100644 --- a/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java +++ b/server/src/test/java/org/apache/druid/discovery/BaseNodeRoleWatcherTest.java @@ -319,72 +319,6 @@ public void testChildRemovedIfPresentRepeatedRemovalsAreIdempotent() Assertions.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList); } - @Test(timeout = 60_000L) - public void testChildRemovedIfPresentRemovesKnownNode() - { - BaseNodeRoleWatcher nodeRoleWatcher = BaseNodeRoleWatcher.create(exec, NodeRole.BROKER); - - DiscoveryDruidNode broker1 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker1"); - - nodeRoleWatcher.childAdded(broker1); - nodeRoleWatcher.cacheInitialized(); - - TestListener listener = new TestListener(); - nodeRoleWatcher.registerListener(listener); - - Assert.assertEquals(ImmutableList.of(broker1), listener.nodesAddedList); - - // Remove with skipIfUnknown=true — node IS in cache, should remove and notify - nodeRoleWatcher.childRemoved(broker1, true); - - Assert.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList); - Assert.assertTrue(nodeRoleWatcher.getAllNodes().isEmpty()); - } - - @Test(timeout = 60_000L) - public void testChildRemovedIfPresentSkipsUnknownNode() - { - BaseNodeRoleWatcher nodeRoleWatcher = BaseNodeRoleWatcher.create(exec, NodeRole.BROKER); - - DiscoveryDruidNode broker1 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker1"); - DiscoveryDruidNode broker2 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker2"); - - nodeRoleWatcher.childAdded(broker1); - nodeRoleWatcher.cacheInitialized(); - - TestListener listener = new TestListener(); - nodeRoleWatcher.registerListener(listener); - - // Remove broker2 with skipIfUnknown=true — node is NOT in cache, should silently skip - nodeRoleWatcher.childRemoved(broker2, true); - - Assert.assertTrue(listener.nodesRemovedList.isEmpty()); - Assert.assertEquals(1, nodeRoleWatcher.getAllNodes().size()); - } - - @Test(timeout = 60_000L) - public void testChildRemovedIfPresentRepeatedRemovalsAreIdempotent() - { - BaseNodeRoleWatcher nodeRoleWatcher = BaseNodeRoleWatcher.create(exec, NodeRole.BROKER); - - DiscoveryDruidNode broker1 = buildDiscoveryDruidNode(NodeRole.BROKER, "broker1"); - - nodeRoleWatcher.childAdded(broker1); - nodeRoleWatcher.cacheInitialized(); - - TestListener listener = new TestListener(); - nodeRoleWatcher.registerListener(listener); - - // First removal should remove and notify - nodeRoleWatcher.childRemoved(broker1, true); - Assert.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList); - - // Second removal should silently skip (node already removed) - nodeRoleWatcher.childRemoved(broker1, true); - // Still only one removal notification - Assert.assertEquals(ImmutableList.of(broker1), listener.nodesRemovedList); - } - private DiscoveryDruidNode buildDiscoveryDruidNode(NodeRole role, String host) { return new DiscoveryDruidNode( From c66ddf89c9ead1eca8e6f2bde0a6c6a8410bb7e0 Mon Sep 17 00:00:00 2001 From: Vivek Dhiman Date: Mon, 7 Sep 2026 11:59:35 +0530 Subject: [PATCH 7/7] Handle Bookmark event Address review comments Signed-off-by: Vivek Dhiman --- extensions-core/kubernetes-extensions/pom.xml | 6 + .../k8s/discovery/DefaultK8sApiClient.java | 10 +- .../K8sDruidNodeDiscoveryProvider.java | 7 +- .../DefaultK8sApiClientWatchPodsTest.java | 164 ++++++++++++++++++ .../K8sDruidNodeDiscoveryProviderTest.java | 51 ++++++ 5 files changed, 234 insertions(+), 4 deletions(-) create mode 100644 extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientWatchPodsTest.java diff --git a/extensions-core/kubernetes-extensions/pom.xml b/extensions-core/kubernetes-extensions/pom.xml index 7decd1a881f7..b1340affb034 100644 --- a/extensions-core/kubernetes-extensions/pom.xml +++ b/extensions-core/kubernetes-extensions/pom.xml @@ -111,6 +111,12 @@ easymock test + + com.squareup.okhttp3 + mockwebserver3 + ${okhttp.version} + test + diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java index f9c30e2e4ad0..95c8dfd4154b 100644 --- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java +++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java @@ -264,8 +264,14 @@ public boolean hasNext() throws IOException return true; } else if (item != null && item.type != null && item.type.equals(WatchResult.BOOKMARK)) { // Events with type BOOKMARK will only contain resourceVersion and no metadata. See - // Kubernetes API documentation for details. - LOGGER.debug("BOOKMARK event fired, no nothing, only update resourceVersion"); + // Kubernetes API documentation for details. Build a response carrying just the + //updated resourceVersion; otherwise obj would still hold the previous event (or + // null on the first event), corrupting the caller's resourceVersion tracking. + LOGGER.debug("BOOKMARK event fired, only updating resourceVersion"); + obj = new Watch.Response<>( + WatchResult.BOOKMARK, + new DiscoveryDruidNodeAndResourceVersion( + item.object.getMetadata().getResourceVersion(), null)); return true; } else { LOGGER.error("WTH! item or item.type is NULL"); 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 bccf5eae2c0e..522b7aa42417 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 @@ -255,8 +255,7 @@ 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)) { if (iter == null) { // history not available, we need to start from scratch @@ -295,6 +294,10 @@ private void keepWatching(String labelSelector, String resourceVersion) } catch (ChannelResetException ex) { LOGGER.debug("Watch stream terminated normally for role[%s], restarting", this.nodeRole); + // Apply the same backoff as other retry paths below: a persistently reset stream + // (flaky proxy, LB idle timeout shorter than expected) would otherwise cause an + // unbounded relist/watch loop against the API server. + sleep(watcherErrorRetryWaitMS); return; } catch (SocketTimeoutException ex) { diff --git a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientWatchPodsTest.java b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientWatchPodsTest.java new file mode 100644 index 000000000000..33a73d972e32 --- /dev/null +++ b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientWatchPodsTest.java @@ -0,0 +1,164 @@ +/* + * 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.druid.k8s.discovery; + +import com.fasterxml.jackson.databind.InjectableValues; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.kubernetes.client.openapi.ApiClient; +import io.kubernetes.client.util.Watch; +import mockwebserver3.MockResponse; +import mockwebserver3.MockWebServer; +import mockwebserver3.SocketEffect; +import okhttp3.Headers; +import org.apache.druid.discovery.DiscoveryDruidNode; +import org.apache.druid.discovery.NodeRole; +import org.apache.druid.jackson.DefaultObjectMapper; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.server.DruidNode; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; + +/** + * Exercises {@link DefaultK8sApiClient#watchPods} end to end against a real (but local, canned) + * HTTP watch stream: BOOKMARK events must surface their resourceVersion to the caller instead of + * a stale or null response, and a mid-stream connection reset must surface as + * {@link ChannelResetException} rather than an unrelated exception type. + */ +public class DefaultK8sApiClientWatchPodsTest +{ + private static final ObjectMapper JSON_MAPPER = new DefaultObjectMapper(); + + static { + // DiscoveryDruidNode's @JsonCreator takes a @JacksonInject ObjectMapper, so the mapper + // used to deserialize it must know how to inject itself. + JSON_MAPPER.setInjectableValues(new InjectableValues.Std().addValue(ObjectMapper.class, JSON_MAPPER)); + } + + private final DiscoveryDruidNode testNode = new DiscoveryDruidNode( + new DruidNode("druid/router", "test-host", true, 80, null, true, false), NodeRole.ROUTER, null); + + private MockWebServer server; + private DefaultK8sApiClient client; + + @BeforeEach + public void setUp() throws IOException + { + server = new MockWebServer(); + server.start(); + + ApiClient apiClient = new ApiClient().setBasePath(server.url("/").toString()); + client = new DefaultK8sApiClient(apiClient, JSON_MAPPER); + } + + @AfterEach + public void tearDown() throws IOException + { + server.close(); + } + + @Test + public void testBookmarkAsFirstEventSurfacesResourceVersion() throws Exception + { + enqueueWatchStream(bookmarkLine("100")); + + try (WatchResult result = client.watchPods("testns", null, "99", NodeRole.ROUTER)) { + Assertions.assertTrue(result.hasNext()); + Watch.Response response = result.next(); + + Assertions.assertNotNull(response, "BOOKMARK event must not surface as a null response"); + Assertions.assertEquals(WatchResult.BOOKMARK, response.type); + Assertions.assertNotNull(response.object); + Assertions.assertEquals("100", response.object.getResourceVersion()); + } + } + + @Test + public void testBookmarkAfterRealEventSurfacesItsOwnResourceVersion() throws Exception + { + enqueueWatchStream(addedLine("200", testNode) + bookmarkLine("201")); + + try (WatchResult result = client.watchPods("testns", null, "99", NodeRole.ROUTER)) { + Assertions.assertTrue(result.hasNext()); + Watch.Response added = result.next(); + Assertions.assertEquals(WatchResult.ADDED, added.type); + Assertions.assertEquals("200", added.object.getResourceVersion()); + + Assertions.assertTrue(result.hasNext()); + Watch.Response bookmark = result.next(); + + Assertions.assertNotNull(bookmark, "BOOKMARK event must not surface as a null response"); + Assertions.assertEquals(WatchResult.BOOKMARK, bookmark.type); + Assertions.assertNotNull(bookmark.object); + Assertions.assertEquals( + "201", + bookmark.object.getResourceVersion(), + "BOOKMARK must carry its own resourceVersion, not the preceding event's"); + } + } + + @Test + public void testConnectionResetMidStreamSurfacesAsChannelResetException() throws Exception + { + // Truncate the body mid-object and sever the connection while it's being read, simulating + // a server-side stream reset. okhttp surfaces this as a ProtocolException, which + // DefaultK8sApiClient must remap to ChannelResetException rather than an unrelated + // RuntimeException, so callers can distinguish it from other IO failures. + server.enqueue(new MockResponse.Builder() + .code(200) + .headers(new Headers.Builder() + .add("Content-Type", "application/json") + .build()) + .body(addedLine("200", testNode).substring(0, 20)) + .onResponseBody(new SocketEffect.CloseSocket(true, true, true)) + .build()); + + try (WatchResult result = client.watchPods("testns", null, "99", NodeRole.ROUTER)) { + Assertions.assertThrows(ChannelResetException.class, result::hasNext); + } + } + + private void enqueueWatchStream(String body) + { + server.enqueue(new MockResponse( + 200, + new Headers.Builder().add("Content-Type", "application/json").build(), + body)); + } + + private static String bookmarkLine(String resourceVersion) + { + return "{\"type\":\"BOOKMARK\",\"object\":{\"kind\":\"Pod\",\"apiVersion\":\"v1\",\"metadata\":{\"resourceVersion\":\"" + + resourceVersion + "\"}}}\n"; + } + + private static String addedLine(String resourceVersion, DiscoveryDruidNode node) throws Exception + { + String infoAnnotation = K8sDruidNodeAnnouncer.getInfoAnnotation(node.getNodeRole()); + String nodeJson = StringUtils.replace(JSON_MAPPER.writeValueAsString(node), "\"", "\\\""); + return "{\"type\":\"ADDED\",\"object\":{\"kind\":\"Pod\",\"apiVersion\":\"v1\",\"metadata\":{" + + "\"name\":\"test-pod\",\"resourceVersion\":\"" + resourceVersion + "\"," + + "\"annotations\":{\"" + infoAnnotation + "\":\"" + nodeJson + "\"}}," + + "\"status\":{\"containerStatuses\":[{\"name\":\"main\",\"ready\":true}]}}}\n"; + } +} 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 a1a8b57ae0b7..445247575fa8 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 @@ -552,6 +552,57 @@ public void testNodeRoleWatcherHandlesChannelResetException() throws Exception discoveryProvider.stop(); } + @Test + @Timeout(value = 10_000, unit = TimeUnit.MILLISECONDS) + public void testPersistentChannelResetAppliesBackoffBetweenRelists() throws Exception + { + String labelSelector = + "druidDiscoveryAnnouncement-cluster-identifier=druid-cluster,druidDiscoveryAnnouncement-router=true"; + K8sApiClient mockK8sApiClient = EasyMock.createMock(K8sApiClient.class); + + // Three consecutive watch attempts are immediately reset, followed by a clean one. Without + // backoff on the ChannelResetException path, these three relist/watch cycles would complete + // almost instantly; with backoff, each reset costs watcherErrorRetryWaitMS. + for (int i = 0; i < 3; i++) { + EasyMock.expect(mockK8sApiClient.listPods(podInfo.getPodNamespace(), labelSelector, NodeRole.ROUTER)) + .andReturn(new DiscoveryDruidNodeList("v1", ImmutableMap.of())); + EasyMock.expect(mockK8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector, "v1", NodeRole.ROUTER)) + .andReturn(new MockWatchResult(Collections.emptyList(), false, false, true)); + } + + EasyMock.expect(mockK8sApiClient.listPods(podInfo.getPodNamespace(), labelSelector, NodeRole.ROUTER)) + .andReturn(new DiscoveryDruidNodeList( + "v2", ImmutableMap.of(testNode1.getDruidNode().getHostAndPortToUse(), testNode1))); + EasyMock.expect(mockK8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector, "v2", NodeRole.ROUTER)) + .andReturn(new MockWatchResult(Collections.emptyList(), false, false)); + + EasyMock.replay(mockK8sApiClient); + + long watcherErrorRetryWaitMS = 1500; + K8sDruidNodeDiscoveryProvider discoveryProvider = + new K8sDruidNodeDiscoveryProvider(podInfo, discoveryConfig, mockK8sApiClient, watcherErrorRetryWaitMS); + discoveryProvider.start(); + + K8sDruidNodeDiscoveryProvider.NodeRoleWatcher nodeDiscovery = + discoveryProvider.getForNodeRole(NodeRole.ROUTER, false); + + MockListener testListener = + new MockListener(ImmutableList.of(MockListener.Event.inited(), MockListener.Event.added(testNode1))); + nodeDiscovery.registerListener(testListener); + + long start = System.currentTimeMillis(); + nodeDiscovery.start(); + testListener.assertSuccess(); + long elapsed = System.currentTimeMillis() - start; + + discoveryProvider.stop(); + + Assertions.assertTrue( + elapsed >= 3 * watcherErrorRetryWaitMS, + "Expected at least 3 backoff waits (" + (3 * watcherErrorRetryWaitMS) + "ms) for 3 persistent " + + "channel resets, but only " + elapsed + "ms elapsed"); + } + private static class MockListener implements DruidNodeDiscovery.Listener { List events;