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 4a4c8838da5 minor: Add QOS filtering for coordinator endpoints (#19271)
4a4c8838da5 is described below
commit 4a4c8838da5b07fdf4563d65db02f9464dc17e23
Author: Karan Kumar <[email protected]>
AuthorDate: Wed Jul 22 23:50:39 2026 +0530
minor: Add QOS filtering for coordinator endpoints (#19271)
---
docs/configuration/index.md | 2 +-
.../supervisor/SeekableStreamSupervisor.java | 12 +-
.../druid/server/initialization/ServerConfig.java | 12 ++
.../jetty/CliIndexerServerModule.java | 8 +-
.../server/initialization/jetty/JettyBindings.java | 41 +++++-
.../jetty/PathExcludingQoSFilter.java | 82 +++++++++++
.../jetty/PathExcludingQoSFilterTest.java | 154 +++++++++++++++++++++
.../java/org/apache/druid/cli/CliCoordinator.java | 20 +++
.../java/org/apache/druid/cli/CliOverlord.java | 32 ++---
.../org/apache/druid/cli/CliCoordinatorTest.java | 121 ++++++++++++++++
.../java/org/apache/druid/cli/CliOverlordTest.java | 23 +--
website/.spelling | 1 +
12 files changed, 453 insertions(+), 55 deletions(-)
diff --git a/docs/configuration/index.md b/docs/configuration/index.md
index 308b523a0da..f4c40ba15b7 100644
--- a/docs/configuration/index.md
+++ b/docs/configuration/index.md
@@ -739,7 +739,7 @@ These Coordinator static configurations can be defined in
the `coordinator/runti
|`druid.coordinator.kill.bufferPeriod`|The amount of time that a segment must
be unused before it is able to be permanently removed from metadata and deep
storage. This can serve as a buffer period to prevent data loss if data ends up
being needed after being marked unused.|`P30D`|
|`druid.coordinator.kill.maxSegments`|The number of unused segments to kill
per kill task. This number must be greater than 0. This only applies when
`druid.coordinator.kill.on=true`.|100|
|`druid.coordinator.kill.maxInterval`|The largest interval, as an [ISO 8601
duration](https://en.wikipedia.org/wiki/ISO_8601#Durations), of segments to
delete per kill task. Set to zero, e.g. `PT0S`, for unlimited. This only
applies when `druid.coordinator.kill.on=true`.|`P30D`|
-
+|`druid.coordinator.server.maxConcurrentRequests`|Maximum number of requests
to non-exempt Coordinator API paths processed concurrently. Requests beyond
this are queued briefly and rejected with HTTP 503 if no slot frees. Leadership
endpoints (`/leader`, `/isLeader`) and non-matching paths such as `/status/*`
are never throttled. This bounds Jetty thread-pool exhaustion from heavy or
excessive calls (for example large `/loadstatus`, `/metadata/segments`, or
`/intervals` scans). Note: thi [...]
##### Metadata management
|Property|Description|Required|Default|
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
index 5b6220fb246..e2a5a4b1ccd 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
@@ -1456,12 +1456,12 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
// from actively reading to pending completion, etc.
// This is a potential silent-loss window where data will not be
queryable until handoff.
log.info(
- "Could not find any task matching taskAllocatorId[%s] in
supervisor[%s] for upgraded pending segment[%s]"
- + " (upgradedFrom[%s]); it will not be re-announced until handoff.",
- taskAllocatorId,
- supervisorId,
- pendingSegmentRecord.getId(),
- pendingSegmentRecord.getUpgradedFromSegmentId()
+ "Could not find any task matching taskAllocatorId[%s] in
supervisor[%s] for upgraded pending segment[%s]"
+ + " (upgradedFrom[%s]); it will not be re-announced
until handoff.",
+ taskAllocatorId,
+ supervisorId,
+ pendingSegmentRecord.getId(),
+ pendingSegmentRecord.getUpgradedFromSegmentId()
);
emitter.emit(
IndexTaskUtils.setPendingSegmentDimensions(getMetricBuilder(),
pendingSegmentRecord)
diff --git
a/server/src/main/java/org/apache/druid/server/initialization/ServerConfig.java
b/server/src/main/java/org/apache/druid/server/initialization/ServerConfig.java
index a7206ca9cf7..2eff2f54a32 100644
---
a/server/src/main/java/org/apache/druid/server/initialization/ServerConfig.java
+++
b/server/src/main/java/org/apache/druid/server/initialization/ServerConfig.java
@@ -47,6 +47,7 @@ import javax.validation.constraints.NotNull;
import java.io.IOException;
import java.util.List;
import java.util.Objects;
+import java.util.Properties;
import java.util.concurrent.TimeUnit;
import java.util.zip.Deflater;
@@ -441,6 +442,17 @@ public class ServerConfig
return Math.max(10, (JvmUtils.getRuntimeInfo().getAvailableProcessors() *
17) / 16 + 2) + 30;
}
+ public static int getNumThreadsFromProperties(Properties properties)
+ {
+ final String value =
properties.getProperty("druid.server.http.numThreads");
+ return value == null ? getDefaultNumThreads() : Integer.parseInt(value);
+ }
+
+ public static int getDefaultMaxConcurrentRequests(int numThreads)
+ {
+ return Math.max(1, Math.max(numThreads - 4, (int) (numThreads * 0.8)));
+ }
+
public static class UriComplianceDeserializer extends
JsonDeserializer<UriCompliance>
{
@Override
diff --git
a/server/src/main/java/org/apache/druid/server/initialization/jetty/CliIndexerServerModule.java
b/server/src/main/java/org/apache/druid/server/initialization/jetty/CliIndexerServerModule.java
index c1c10547d88..f8768f3a7f0 100644
---
a/server/src/main/java/org/apache/druid/server/initialization/jetty/CliIndexerServerModule.java
+++
b/server/src/main/java/org/apache/druid/server/initialization/jetty/CliIndexerServerModule.java
@@ -47,7 +47,6 @@ import java.util.Properties;
*/
public class CliIndexerServerModule implements Module
{
- public static final String SERVER_HTTP_NUM_THREADS_PROPERTY =
"druid.server.http.numThreads";
private final Properties properties;
public CliIndexerServerModule(Properties properties)
@@ -62,12 +61,7 @@ public class CliIndexerServerModule implements Module
LifecycleModule.register(binder, ChatHandlerResource.class);
// Use an equal number of threads for chat handler and non-chat handler
requests.
- int serverHttpNumThreads;
- if (properties.getProperty(SERVER_HTTP_NUM_THREADS_PROPERTY) == null) {
- serverHttpNumThreads = ServerConfig.getDefaultNumThreads();
- } else {
- serverHttpNumThreads =
Integer.parseInt(properties.getProperty(SERVER_HTTP_NUM_THREADS_PROPERTY));
- }
+ int serverHttpNumThreads =
ServerConfig.getNumThreadsFromProperties(properties);
JettyBindings.addQosFilter(
binder,
diff --git
a/server/src/main/java/org/apache/druid/server/initialization/jetty/JettyBindings.java
b/server/src/main/java/org/apache/druid/server/initialization/jetty/JettyBindings.java
index 3934e33277b..2c0ed518e0a 100644
---
a/server/src/main/java/org/apache/druid/server/initialization/jetty/JettyBindings.java
+++
b/server/src/main/java/org/apache/druid/server/initialization/jetty/JettyBindings.java
@@ -22,7 +22,6 @@ package org.apache.druid.server.initialization.jetty;
import com.google.common.collect.ImmutableMap;
import com.google.inject.Binder;
import com.google.inject.multibindings.Multibinder;
-import org.eclipse.jetty.ee8.servlets.QoSFilter;
import org.eclipse.jetty.server.Handler;
import javax.servlet.DispatcherType;
@@ -43,6 +42,18 @@ public class JettyBindings
}
public static void addQosFilter(Binder binder, String[] paths, int
maxRequests)
+ {
+ addQosFilter(binder, paths, maxRequests, null);
+ }
+
+ /**
+ * Registers a QoS filter for the given {@code paths}, exempting any request
+ * that matches one of {@code excludedPaths} (servlet path-specs) from QoS
+ * throttling. Exclusions are needed because servlet filter mappings cannot
+ * express a "match this prefix except this sub-path" rule; see
+ * {@link PathExcludingQoSFilter}.
+ */
+ public static void addQosFilter(Binder binder, String[] paths, int
maxRequests, String[] excludedPaths)
{
if (maxRequests <= 0) {
return;
@@ -50,7 +61,7 @@ public class JettyBindings
Multibinder.newSetBinder(binder, QosFilterHolder.class)
.addBinding()
- .toInstance(new QosFilterHolder(paths, maxRequests));
+ .toInstance(new QosFilterHolder(paths, maxRequests,
excludedPaths));
}
public static void addHandler(Binder binder, Class<? extends Handler>
handlerClass)
@@ -67,28 +78,46 @@ public class JettyBindings
private final long timeoutMs;
- public QosFilterHolder(String[] paths, int maxRequests, long timeoutMs)
+ private final String[] excludedPaths;
+
+ public QosFilterHolder(String[] paths, int maxRequests, long timeoutMs,
String[] excludedPaths)
{
this.paths = paths;
this.maxRequests = maxRequests;
this.timeoutMs = timeoutMs;
+ this.excludedPaths = excludedPaths == null ? new String[0] :
excludedPaths;
+ }
+
+ public QosFilterHolder(String[] paths, int maxRequests, long timeoutMs)
+ {
+ this(paths, maxRequests, timeoutMs, null);
+ }
+
+ public QosFilterHolder(String[] paths, int maxRequests, String[]
excludedPaths)
+ {
+ this(paths, maxRequests, -1, excludedPaths);
}
public QosFilterHolder(String[] paths, int maxRequests)
{
- this(paths, maxRequests, -1);
+ this(paths, maxRequests, -1, null);
}
@Override
public Filter getFilter()
{
- return new QoSFilter();
+ return new PathExcludingQoSFilter(excludedPaths);
}
@Override
public Class<? extends Filter> getFilterClass()
{
- return QoSFilter.class;
+ return PathExcludingQoSFilter.class;
+ }
+
+ public String[] getExcludedPaths()
+ {
+ return excludedPaths;
}
@Override
diff --git
a/server/src/main/java/org/apache/druid/server/initialization/jetty/PathExcludingQoSFilter.java
b/server/src/main/java/org/apache/druid/server/initialization/jetty/PathExcludingQoSFilter.java
new file mode 100644
index 00000000000..e5b5066dd0f
--- /dev/null
+++
b/server/src/main/java/org/apache/druid/server/initialization/jetty/PathExcludingQoSFilter.java
@@ -0,0 +1,82 @@
+/*
+ * 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.server.initialization.jetty;
+
+import org.eclipse.jetty.ee8.servlets.QoSFilter;
+import org.eclipse.jetty.http.pathmap.PathSpecSet;
+
+import javax.servlet.FilterChain;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletRequest;
+import java.io.IOException;
+
+/**
+ * A {@link QoSFilter} that allows a set of request paths to bypass QoS
+ * throttling entirely. This is useful for lightweight leadership /
health-check
+ * endpoints (such as {@code /druid/coordinator/v1/isLeader}) that must remain
+ * responsive even when heavier APIs matched by the same broad filter path-spec
+ * have saturated the QoS semaphore. Servlet filter mappings cannot express
path
+ * exclusions, so the exclusion is applied here at request time.
+ *
+ * <p>Excluded requests are passed straight down the filter chain without ever
+ * acquiring the QoS semaphore; all other requests are handled by the standard
+ * {@link QoSFilter} behavior.
+ */
+public class PathExcludingQoSFilter extends QoSFilter
+{
+ private final PathSpecSet excludedPaths;
+
+ public PathExcludingQoSFilter(String[] excludedPaths)
+ {
+ this.excludedPaths = new PathSpecSet();
+ if (excludedPaths != null) {
+ for (String path : excludedPaths) {
+ this.excludedPaths.add(path);
+ }
+ }
+ }
+
+ @Override
+ public void doFilter(ServletRequest request, ServletResponse response,
FilterChain chain)
+ throws IOException, ServletException
+ {
+ if (isExcluded(request)) {
+ chain.doFilter(request, response);
+ } else {
+ super.doFilter(request, response, chain);
+ }
+ }
+
+ private boolean isExcluded(ServletRequest request)
+ {
+ if (!(request instanceof HttpServletRequest)) {
+ return false;
+ }
+ final HttpServletRequest httpRequest = (HttpServletRequest) request;
+ final String contextPath = httpRequest.getContextPath();
+ String path = httpRequest.getRequestURI();
+ if (contextPath != null && !contextPath.isEmpty() &&
path.startsWith(contextPath)) {
+ path = path.substring(contextPath.length());
+ }
+ return excludedPaths.test(path);
+ }
+}
diff --git
a/server/src/test/java/org/apache/druid/server/initialization/jetty/PathExcludingQoSFilterTest.java
b/server/src/test/java/org/apache/druid/server/initialization/jetty/PathExcludingQoSFilterTest.java
new file mode 100644
index 00000000000..3fe56d5053a
--- /dev/null
+++
b/server/src/test/java/org/apache/druid/server/initialization/jetty/PathExcludingQoSFilterTest.java
@@ -0,0 +1,154 @@
+/*
+ * 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.server.initialization.jetty;
+
+import org.apache.druid.server.mocks.MockHttpServletRequest;
+import org.apache.druid.server.mocks.MockHttpServletResponse;
+import org.junit.Assert;
+import org.junit.Test;
+
+import javax.servlet.FilterChain;
+import javax.servlet.FilterConfig;
+import javax.servlet.ServletContext;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletResponse;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.Map;
+
+public class PathExcludingQoSFilterTest
+{
+ // The concrete request URIs those specs are expected to exempt.
+ private static final String[] EXCLUDED_PATHS = new String[]{
+ "/druid/coordinator/v1/isLeader",
+ "/druid/coordinator/v1/leader"
+ };
+
+ @Test
+ public void testExcludedPathsBypassQosFilter() throws Exception
+ {
+ // The filter is intentionally NOT initialized: excluded requests must
never touch the QoS semaphore,
+ // so if super.doFilter() were invoked it would fail on the uninitialized
semaphore.
+ final PathExcludingQoSFilter filter = new
PathExcludingQoSFilter(EXCLUDED_PATHS);
+
+ for (String excludedPath : EXCLUDED_PATHS) {
+ final MockHttpServletRequest request = request(excludedPath);
+ final MockHttpServletResponse response = new MockHttpServletResponse();
+ final RecordingFilterChain chain = new RecordingFilterChain();
+
+ filter.doFilter(request, response, chain);
+
+ Assert.assertEquals("Excluded request should be passed straight down the
chain", 1, chain.invocations);
+ Assert.assertSame(request, chain.lastRequest);
+ Assert.assertEquals("Excluded request should not be rejected", 0,
response.getStatus());
+ }
+ }
+
+ @Test
+ public void testNonExcludedPathIsHandledByQosFilter() throws Exception
+ {
+ final PathExcludingQoSFilter filter = new
PathExcludingQoSFilter(EXCLUDED_PATHS);
+ filter.init(new DummyFilterConfig(Map.of("maxRequests", "1")));
+
+ final MockHttpServletRequest request =
request("/druid/coordinator/v1/loadstatus");
+ final MockHttpServletResponse response = new MockHttpServletResponse();
+ final RecordingFilterChain chain = new RecordingFilterChain();
+
+ // A single request with a free semaphore is accepted and passed down the
chain by the standard QoSFilter.
+ filter.doFilter(request, response, chain);
+
+ Assert.assertEquals("Non-excluded request should be handled by the QoS
filter", 1, chain.invocations);
+ Assert.assertNotEquals(
+ "Accepted request must not be rejected with 503",
+ HttpServletResponse.SC_SERVICE_UNAVAILABLE,
+ response.getStatus()
+ );
+ }
+
+ @Test
+ public void testNullExcludedPathsHandlesAllRequests()
+ {
+ // A null exclusion list must not blow up; nothing should be treated as
excluded.
+ final PathExcludingQoSFilter filter = new PathExcludingQoSFilter(null);
+ Assert.assertNotNull(filter);
+ }
+
+ private static MockHttpServletRequest request(String requestUri)
+ {
+ final MockHttpServletRequest request = new MockHttpServletRequest()
+ {
+ @Override
+ public String getContextPath()
+ {
+ return "";
+ }
+ };
+ request.requestUri = requestUri;
+ return request;
+ }
+
+ private static class RecordingFilterChain implements FilterChain
+ {
+ private int invocations = 0;
+ private ServletRequest lastRequest;
+
+ @Override
+ public void doFilter(ServletRequest request, ServletResponse response)
+ {
+ invocations++;
+ lastRequest = request;
+ }
+ }
+
+ private static class DummyFilterConfig implements FilterConfig
+ {
+ private final Map<String, String> initParameters;
+
+ DummyFilterConfig(Map<String, String> initParameters)
+ {
+ this.initParameters = initParameters;
+ }
+
+ @Override
+ public String getFilterName()
+ {
+ return "qos";
+ }
+
+ @Override
+ public ServletContext getServletContext()
+ {
+ return null;
+ }
+
+ @Override
+ public String getInitParameter(String name)
+ {
+ return initParameters.get(name);
+ }
+
+ @Override
+ public Enumeration<String> getInitParameterNames()
+ {
+ return Collections.enumeration(initParameters.keySet());
+ }
+ }
+}
diff --git a/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
b/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
index f8b52fafbdf..7fc344d781d 100644
--- a/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
+++ b/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
@@ -105,6 +105,8 @@ import org.apache.druid.server.http.RulesResource;
import org.apache.druid.server.http.SelfDiscoveryResource;
import org.apache.druid.server.http.ServersResource;
import org.apache.druid.server.http.TiersResource;
+import org.apache.druid.server.initialization.ServerConfig;
+import org.apache.druid.server.initialization.jetty.JettyBindings;
import org.apache.druid.server.initialization.jetty.JettyServerInitializer;
import org.apache.druid.server.lookup.cache.LookupCoordinatorManager;
import org.apache.druid.server.lookup.cache.LookupCoordinatorManagerConfig;
@@ -231,6 +233,24 @@ public class CliCoordinator extends ServerRunnable
binder.bind(JettyServerInitializer.class)
.to(CoordinatorJettyServerInitializer.class);
+ // QoS filtering to prevent heavy coordinator API requests from
starving health check endpoints.
+ // Set druid.coordinator.server.maxConcurrentRequests=-1 to
disable.
+ final int serverHttpNumThreads =
ServerConfig.getNumThreadsFromProperties(properties);
+ final int maxConcurrentRequests =
properties.containsKey("druid.coordinator.server.maxConcurrentRequests")
+ ?
Integer.parseInt(properties.getProperty("druid.coordinator.server.maxConcurrentRequests"))
+ :
ServerConfig.getDefaultMaxConcurrentRequests(serverHttpNumThreads);
+ if (maxConcurrentRequests > 0) {
+ log.info("Coordinator QoS filtering enabled. Max concurrent
requests: [%d]", maxConcurrentRequests);
+ JettyBindings.addQosFilter(
+ binder,
+ new String[]{"/druid/coordinator/v1/*", "/druid-internal/*"},
+ maxConcurrentRequests,
+ new String[]{"/druid/coordinator/v1/isLeader",
"/druid/coordinator/v1/leader"}
+ );
+ } else {
+ log.info("Coordinator QoS filtering disabled.");
+ }
+
Jerseys.addResource(binder, CoordinatorResource.class);
Jerseys.addResource(binder, CoordinatorCompactionResource.class);
Jerseys.addResource(binder,
CoordinatorDynamicConfigsResource.class);
diff --git a/services/src/main/java/org/apache/druid/cli/CliOverlord.java
b/services/src/main/java/org/apache/druid/cli/CliOverlord.java
index 67a8be584df..50116cb4968 100644
--- a/services/src/main/java/org/apache/druid/cli/CliOverlord.java
+++ b/services/src/main/java/org/apache/druid/cli/CliOverlord.java
@@ -129,7 +129,6 @@ import org.apache.druid.server.http.RedirectFilter;
import org.apache.druid.server.http.RedirectInfo;
import org.apache.druid.server.http.SelfDiscoveryResource;
import org.apache.druid.server.initialization.ServerConfig;
-import org.apache.druid.server.initialization.jetty.CliIndexerServerModule;
import org.apache.druid.server.initialization.jetty.JettyBindings;
import org.apache.druid.server.initialization.jetty.JettyServerInitUtils;
import org.apache.druid.server.initialization.jetty.JettyServerInitializer;
@@ -483,27 +482,17 @@ public class CliOverlord extends ServerRunnable
Jerseys.addResource(binder, OverlordDataSourcesResource.class);
- final int serverHttpNumThreads =
properties.containsKey(CliIndexerServerModule.SERVER_HTTP_NUM_THREADS_PROPERTY)
- ?
Integer.parseInt(properties.getProperty(CliIndexerServerModule.SERVER_HTTP_NUM_THREADS_PROPERTY))
- :
ServerConfig.getDefaultNumThreads();
-
- final int maxConcurrentActions;
- if
(properties.containsKey("druid.indexer.server.maxConcurrentActions")) {
- maxConcurrentActions =
Integer.parseInt(properties.getProperty("druid.indexer.server.maxConcurrentActions"));
- } else {
- maxConcurrentActions =
getDefaultMaxConcurrentActions(serverHttpNumThreads);
- }
-
+ // QoS filtering to prevent action requests from starving health
check endpoints.
+ // Set druid.indexer.server.maxConcurrentActions=-1 to disable.
+ final int serverHttpNumThreads =
ServerConfig.getNumThreadsFromProperties(properties);
+ final int maxConcurrentActions =
properties.containsKey("druid.indexer.server.maxConcurrentActions")
+ ?
Integer.parseInt(properties.getProperty("druid.indexer.server.maxConcurrentActions"))
+ :
ServerConfig.getDefaultMaxConcurrentRequests(serverHttpNumThreads);
if (maxConcurrentActions > 0) {
- // Add QoS filtering for action endpoints only
- final String[] actionPaths = {
- "/druid/indexer/v1/action",
- };
-
log.info("Overlord QoS filtering enabled for action endpoints.
Max concurrent actions: [%d]", maxConcurrentActions);
- JettyBindings.addQosFilter(binder, actionPaths,
maxConcurrentActions);
+ JettyBindings.addQosFilter(binder, "/druid/indexer/v1/action",
maxConcurrentActions);
} else {
- log.info("Overlord QoS filtering disabled for action endpoints.
Max concurrent actions: [%d]", serverHttpNumThreads);
+ log.info("Overlord QoS filtering disabled for action
endpoints.");
}
}
},
@@ -520,11 +509,6 @@ public class CliOverlord extends ServerRunnable
);
}
- public static int getDefaultMaxConcurrentActions(int serverHttpNumThreads)
- {
- return Math.max(1, Math.max(serverHttpNumThreads - 4, (int)
(serverHttpNumThreads * 0.8)));
- }
-
/**
*/
private static class OverlordJettyServerInitializer implements
JettyServerInitializer
diff --git
a/services/src/test/java/org/apache/druid/cli/CliCoordinatorTest.java
b/services/src/test/java/org/apache/druid/cli/CliCoordinatorTest.java
new file mode 100644
index 00000000000..8be519dbd1d
--- /dev/null
+++ b/services/src/test/java/org/apache/druid/cli/CliCoordinatorTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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.cli;
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Key;
+import com.google.inject.Scopes;
+import com.google.inject.TypeLiteral;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.guice.LazySingleton;
+import org.apache.druid.guice.LifecycleModule;
+import org.apache.druid.jackson.JacksonModule;
+import org.apache.druid.server.initialization.jetty.JettyBindings;
+import org.junit.Assert;
+import org.junit.Test;
+
+import javax.validation.Validation;
+import javax.validation.Validator;
+import java.util.Arrays;
+import java.util.Properties;
+import java.util.Set;
+
+public class CliCoordinatorTest
+{
+ private static final String COORDINATOR_QOS_PATH = "/druid/coordinator/v1/*";
+
+ @Test
+ public void testQosFilterIsBoundByDefault()
+ {
+ final Injector injector = makeCoordinatorInjector(new Properties());
+
+ final Set<JettyBindings.QosFilterHolder> qosFilters =
getQosFilterHolders(injector);
+ Assert.assertTrue(
+ "Coordinator QoS filter should be bound when maxConcurrentRequests
defaults to a positive value",
+ hasCoordinatorQosFilter(qosFilters)
+ );
+ }
+
+ @Test
+ public void testQosFilterIsNotBoundWhenDisabled()
+ {
+ final Properties properties = new Properties();
+ properties.setProperty("druid.coordinator.server.maxConcurrentRequests",
"-1");
+ final Injector injector = makeCoordinatorInjector(properties);
+
+ final Set<JettyBindings.QosFilterHolder> qosFilters =
getQosFilterHolders(injector);
+ Assert.assertFalse(
+ "Coordinator QoS filter should not be bound when maxConcurrentRequests
is set to a non-positive value",
+ hasCoordinatorQosFilter(qosFilters)
+ );
+ }
+
+ @Test
+ public void testLeaderEndpointsExcludedFromQos()
+ {
+ final Injector injector = makeCoordinatorInjector(new Properties());
+
+ final JettyBindings.QosFilterHolder coordinatorQosFilter =
+ getQosFilterHolders(injector).stream()
+ .filter(holder ->
Arrays.asList(holder.getPaths()).contains(COORDINATOR_QOS_PATH))
+ .findFirst()
+ .orElseThrow(() -> new
AssertionError("Coordinator QoS filter should be bound"));
+
+
+ final Set<String> excludedPaths =
Set.of(coordinatorQosFilter.getExcludedPaths());
+ Assert.assertTrue(
+ "isLeader should be exempt from QoS filtering",
+ excludedPaths.contains("/druid/coordinator/v1/isLeader")
+ );
+ Assert.assertTrue(
+ "leader should be exempt from QoS filtering",
+ excludedPaths.contains("/druid/coordinator/v1/leader")
+ );
+ }
+
+ private static boolean
hasCoordinatorQosFilter(Set<JettyBindings.QosFilterHolder> qosFilters)
+ {
+ return qosFilters.stream()
+ .anyMatch(holder ->
Arrays.asList(holder.getPaths()).contains(COORDINATOR_QOS_PATH));
+ }
+
+ private static Set<JettyBindings.QosFilterHolder>
getQosFilterHolders(Injector injector)
+ {
+ return injector.getInstance(Key.get(new
TypeLiteral<Set<JettyBindings.QosFilterHolder>>() {}));
+ }
+
+ private static Injector makeCoordinatorInjector(final Properties props)
+ {
+ final Injector baseInjector = Guice.createInjector(
+ new JacksonModule(),
+ new LifecycleModule(),
+ binder -> {
+
binder.bind(Validator.class).toInstance(Validation.buildDefaultValidatorFactory().getValidator());
+ binder.bindScope(LazySingleton.class, Scopes.SINGLETON);
+ binder.bind(Properties.class).toInstance(props);
+ }
+ );
+
+ final CliCoordinator coordinator = new CliCoordinator();
+ baseInjector.injectMembers(coordinator);
+ return coordinator.makeInjector(Set.of(NodeRole.COORDINATOR));
+ }
+}
diff --git a/services/src/test/java/org/apache/druid/cli/CliOverlordTest.java
b/services/src/test/java/org/apache/druid/cli/CliOverlordTest.java
index 031c38ff8b2..9fec28aeeab 100644
--- a/services/src/test/java/org/apache/druid/cli/CliOverlordTest.java
+++ b/services/src/test/java/org/apache/druid/cli/CliOverlordTest.java
@@ -26,6 +26,7 @@ import org.apache.druid.metadata.SegmentsMetadataManager;
import org.apache.druid.metadata.segment.SqlSegmentsMetadataManagerV2;
import org.apache.druid.metadata.segment.cache.HeapMemorySegmentMetadataCache;
import org.apache.druid.metadata.segment.cache.SegmentMetadataCache;
+import org.apache.druid.server.initialization.ServerConfig;
import org.junit.Assert;
import org.junit.Test;
@@ -53,27 +54,27 @@ public class CliOverlordTest
@Test
- public void testGetDefaultMaxConcurrentActions()
+ public void testGetDefaultMaxConcurrentRequests()
{
// Small thread count where
- Assert.assertEquals(8, CliOverlord.getDefaultMaxConcurrentActions(10));
+ Assert.assertEquals(8, ServerConfig.getDefaultMaxConcurrentRequests(10));
// Medium thread count where
- Assert.assertEquals(21, CliOverlord.getDefaultMaxConcurrentActions(25));
- Assert.assertEquals(26, CliOverlord.getDefaultMaxConcurrentActions(30));
+ Assert.assertEquals(21, ServerConfig.getDefaultMaxConcurrentRequests(25));
+ Assert.assertEquals(26, ServerConfig.getDefaultMaxConcurrentRequests(30));
// Large thread count
- Assert.assertEquals(46, CliOverlord.getDefaultMaxConcurrentActions(50));
- Assert.assertEquals(96, CliOverlord.getDefaultMaxConcurrentActions(100));
+ Assert.assertEquals(46, ServerConfig.getDefaultMaxConcurrentRequests(50));
+ Assert.assertEquals(96, ServerConfig.getDefaultMaxConcurrentRequests(100));
// Test edge cases - return atleast 1 thread
- Assert.assertEquals(1, CliOverlord.getDefaultMaxConcurrentActions(-1));
- Assert.assertEquals(1, CliOverlord.getDefaultMaxConcurrentActions(0));
+ Assert.assertEquals(1, ServerConfig.getDefaultMaxConcurrentRequests(-1));
+ Assert.assertEquals(1, ServerConfig.getDefaultMaxConcurrentRequests(0));
// Test small clustesr
- Assert.assertEquals(2, CliOverlord.getDefaultMaxConcurrentActions(3));
- Assert.assertEquals(3, CliOverlord.getDefaultMaxConcurrentActions(4));
- Assert.assertEquals(4, CliOverlord.getDefaultMaxConcurrentActions(5));
+ Assert.assertEquals(2, ServerConfig.getDefaultMaxConcurrentRequests(3));
+ Assert.assertEquals(3, ServerConfig.getDefaultMaxConcurrentRequests(4));
+ Assert.assertEquals(4, ServerConfig.getDefaultMaxConcurrentRequests(5));
}
}
diff --git a/website/.spelling b/website/.spelling
index e1723b254dc..c99c42107f3 100644
--- a/website/.spelling
+++ b/website/.spelling
@@ -1693,6 +1693,7 @@ un
G1GC
GroupBys
QoS-type
+QoS
DumpSegment
SegmentMetadata
__time
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
