This is an automated email from the ASF dual-hosted git repository.
FrankChen021 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 eb43fd0458d fix: Restarted the watch after stream reset #19055 (#19056)
eb43fd0458d is described below
commit eb43fd0458d0f4d3da8cfdebecfcd366d9f45be1
Author: Vivek Dhiman <[email protected]>
AuthorDate: Thu Oct 8 18:28:02 2026 +0530
fix: Restarted the watch after stream reset #19055 (#19056)
* Restarted the watch after stream reset
* Addressed review comments
* Add unit test and addresses styling issues
* Fix service discovery bug in `kubernetes-extensions` (#19139)
* Address review comments.
Remove hard dependency of okhttp3 internals
* 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.
* Handle Bookmark event
Address review comments
Signed-off-by: Vivek Dhiman <[email protected]>
---------
Signed-off-by: Vivek Dhiman <[email protected]>
Co-authored-by: Lucas Capistrant <[email protected]>
---
extensions-core/kubernetes-extensions/pom.xml | 6 +
.../druid/k8s/discovery/ChannelResetException.java | 30 ++++
.../druid/k8s/discovery/DefaultK8sApiClient.java | 19 ++-
.../discovery/K8sDruidNodeDiscoveryProvider.java | 71 ++++-----
.../apache/druid/k8s/discovery/WatchResult.java | 8 +-
.../DefaultK8sApiClientWatchPodsTest.java | 164 +++++++++++++++++++++
.../K8sDruidNodeDiscoveryProviderTest.java | 135 ++++++++++++++++-
7 files changed, 387 insertions(+), 46 deletions(-)
diff --git a/extensions-core/kubernetes-extensions/pom.xml
b/extensions-core/kubernetes-extensions/pom.xml
index 7decd1a881f..b1340affb03 100644
--- a/extensions-core/kubernetes-extensions/pom.xml
+++ b/extensions-core/kubernetes-extensions/pom.xml
@@ -111,6 +111,12 @@
<artifactId>easymock</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.squareup.okhttp3</groupId>
+ <artifactId>mockwebserver3</artifactId>
+ <version>${okhttp.version}</version>
+ <scope>test</scope>
+ </dependency>
<!-- others -->
<dependency>
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 00000000000..345ec04fed9
--- /dev/null
+++
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/ChannelResetException.java
@@ -0,0 +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)
+ {
+ 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 b7992a3fb2a..95c8dfd4154 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
@@ -40,6 +40,7 @@ import org.apache.druid.java.util.common.RE;
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;
@@ -194,7 +195,7 @@ public class DefaultK8sApiClient implements K8sApiClient
private Watch.Response<DiscoveryDruidNodeAndResourceVersion> obj;
@Override
- public boolean hasNext() throws SocketTimeoutException
+ public boolean hasNext() throws IOException
{
try {
while (watch.hasNext()) {
@@ -263,8 +264,14 @@ public class DefaultK8sApiClient implements K8sApiClient
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");
@@ -274,9 +281,11 @@ public class DefaultK8sApiClient implements K8sApiClient
catch (RuntimeException ex) {
if (ex.getCause() instanceof SocketTimeoutException) {
throw (SocketTimeoutException) ex.getCause();
- } else {
- throw ex;
}
+ if (ex.getCause() instanceof IOException && !(ex.getCause()
instanceof InterruptedIOException)) {
+ throw new ChannelResetException(ex.getCause());
+ }
+ throw ex;
}
return false;
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 8dfc595db74..522b7aa4241 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
@@ -254,50 +254,51 @@ public class K8sDruidNodeDiscoveryProvider extends
DruidNodeDiscoveryProvider
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.isStarted()) {
+ 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<DiscoveryDruidNodeAndResourceVersion> 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<DiscoveryDruidNodeAndResourceVersion> 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:
+ 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();
}
+ LOGGER.trace("Watch closed normally for role[%s]", this.nodeRole);
+ return;
+ }
+ 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) {
// 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 2b3ef7b584a..37cac3768c5 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 @@ package org.apache.druid.k8s.discovery;
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
@@ -29,7 +29,7 @@ import java.net.SocketTimeoutException;
* 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";
@@ -44,9 +44,7 @@ public interface WatchResult
*/
String NOT_READY = "NOT_READY";
- boolean hasNext() throws SocketTimeoutException;
+ boolean hasNext() throws IOException;
Watch.Response<DiscoveryDruidNodeAndResourceVersion> next();
-
- void close();
}
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 00000000000..33a73d972e3
--- /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<DiscoveryDruidNodeAndResourceVersion> 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<DiscoveryDruidNodeAndResourceVersion> added =
result.next();
+ Assertions.assertEquals(WatchResult.ADDED, added.type);
+ Assertions.assertEquals("200", added.object.getResourceVersion());
+
+ Assertions.assertTrue(result.hasNext());
+ Watch.Response<DiscoveryDruidNodeAndResourceVersion> 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 6417edb4b27..445247575fa 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.Assertions;
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,122 @@ public class K8sDruidNodeDiscoveryProviderTest
);
}
+ @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();
+ }
+
+ @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<Event> events;
@@ -654,6 +771,7 @@ public class K8sDruidNodeDiscoveryProviderTest
private volatile boolean timeoutOnStart;
private volatile boolean timeooutOnEmptyResults;
+ private volatile boolean channelResetOnStart;
private volatile boolean closeCalled = false;
public MockWatchResult(
@@ -667,13 +785,28 @@ public class K8sDruidNodeDiscoveryProviderTest
this.timeooutOnEmptyResults = timeooutOnEmptyResults;
}
+ public MockWatchResult(
+ List<Watch.Response<DiscoveryDruidNodeAndResourceVersion>> 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) {
+ throw new ChannelResetException(new RuntimeException("simulated
StreamResetException"));
+ }
+
if (results.isEmpty()) {
if (timeooutOnEmptyResults) {
throw new SocketTimeoutException("testing timeout on end!!!");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
