add configs to enable fast request failure on broker and historical

docs/content/configuration/broker.md

@@ -36,7 +36,9 @@ Druid uses Jetty to serve HTTP requests.
 |Property|Description|Default|
 |--------|-----------|-------|
 |`druid.server.http.numThreads`|Number of threads for HTTP requests.|max(10, (Number of cores * 17) / 16 + 2) + 30|
+|`druid.server.http.queueSize`|Size of the worker queue used by Jetty server to temporarily store incoming client connections. If this value is set and a request is rejected by jetty because queue is full then client would observe request failure with TCP connection being closed immediately with a completely empty response from server.|Unbounded|
 |`druid.server.http.maxIdleTime`|The Jetty max idle time for a connection.|PT5m|
+|`druid.server.http.enableRequestLimit`|If enabled, no requests would be queued in jetty queue and "HTTP 429 Too Many Requests" error response would be sent. |false|
 |`druid.server.http.defaultQueryTimeout`|Query timeout in millis, beyond which unfinished queries will be cancelled|300000|
 |`druid.server.http.maxScatterGatherBytes`|Maximum number of bytes gathered from data nodes such as historicals and realtime processes to execute a query. This is an advance configuration that allows to protect in case broker is under heavy load and not utilizing the data gathered in memory fast enough and leading to OOMs. This limit can be further reduced at query time using `maxScatterGatherBytes` in the context. Note that having large limit is not necessarily bad if broker is never under heavy concurrent load in which case data gathered is processed quickly and freeing up the memory used.|Long.MAX_VALUE|
 |`druid.broker.http.numConnections`|Size of connection pool for the Broker to connect to historical and real-time processes. If there are more queries than this number that all need to speak to the same node, then they will queue up.|20|

docs/content/configuration/historical.md

@@ -47,7 +47,9 @@ Druid uses Jetty to serve HTTP requests.
 |Property|Description|Default|
 |--------|-----------|-------|
 |`druid.server.http.numThreads`|Number of threads for HTTP requests.|max(10, (Number of cores * 17) / 16 + 2) + 30|
+|`druid.server.http.queueSize`|Size of the worker queue used by Jetty server to temporarily store incoming client connections. If this value is set and a request is rejected by jetty because queue is full then client would observe request failure with TCP connection being closed immediately with a completely empty response from server.|Unbounded|
 |`druid.server.http.maxIdleTime`|The Jetty max idle time for a connection.|PT5m|
+|`druid.server.http.enableRequestLimit`|If enabled, no requests would be queued in jetty queue and "HTTP 429 Too Many Requests" error response would be sent. |false|
 |`druid.server.http.defaultQueryTimeout`|Query timeout in millis, beyond which unfinished queries will be cancelled|300000|
 
 #### Processing

server/src/main/java/io/druid/server/initialization/ServerConfig.java

@@ -34,6 +34,13 @@ public class ServerConfig
   @Min(1)
   private int numThreads = Math.max(10, (Runtime.getRuntime().availableProcessors() * 17) / 16 + 2) + 30;
 
+  @JsonProperty
+  @Min(1)
+  private int queueSize = Integer.MAX_VALUE;
+
+  @JsonProperty
+  private boolean enableRequestLimit = false;
+
   @JsonProperty
   @NotNull
   private Period maxIdleTime = new Period("PT5m");
@@ -51,6 +58,16 @@ public int getNumThreads()
     return numThreads;
   }
 
+  public int getQueueSize()
+  {
+    return queueSize;
+  }
+
+  public boolean isEnableRequestLimit()
+  {
+    return enableRequestLimit;
+  }
+
   public Period getMaxIdleTime()
   {
     return maxIdleTime;
@@ -77,6 +94,8 @@ public boolean equals(Object o)
     }
     ServerConfig that = (ServerConfig) o;
     return numThreads == that.numThreads &&
+           queueSize == that.queueSize &&
+           enableRequestLimit == that.enableRequestLimit &&
            defaultQueryTimeout == that.defaultQueryTimeout &&
            maxScatterGatherBytes == that.maxScatterGatherBytes &&
            Objects.equals(maxIdleTime, that.maxIdleTime);
@@ -85,17 +104,13 @@ public boolean equals(Object o)
   @Override
   public int hashCode()
   {
-    return Objects.hash(numThreads, maxIdleTime, defaultQueryTimeout, maxScatterGatherBytes);
-  }
-
-  @Override
-  public String toString()
-  {
-    return "ServerConfig{" +
-           "numThreads=" + numThreads +
-           ", maxIdleTime=" + maxIdleTime +
-           ", defaultQueryTimeout=" + defaultQueryTimeout +
-           ", maxScatterGatherBytes=" + maxScatterGatherBytes +
-           '}';
+    return Objects.hash(
+        numThreads,
+        queueSize,
+        enableRequestLimit,
+        maxIdleTime,
+        defaultQueryTimeout,
+        maxScatterGatherBytes
+    );
   }
 }

server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java

@@ -22,16 +22,13 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.jaxrs.json.JacksonJsonProvider;
 import com.fasterxml.jackson.jaxrs.smile.JacksonSmileProvider;
-import com.google.common.collect.Iterables;
 import com.google.common.primitives.Ints;
 import com.google.inject.Binder;
 import com.google.inject.Binding;
-import com.google.inject.ConfigurationException;
 import com.google.inject.Inject;
 import com.google.inject.Injector;
 import com.google.inject.Key;
 import com.google.inject.Provides;
-import com.google.inject.ProvisionException;
 import com.google.inject.Scopes;
 import com.google.inject.Singleton;
 import com.google.inject.multibindings.Multibinder;
@@ -51,6 +48,7 @@
 import io.druid.guice.annotations.Json;
 import io.druid.guice.annotations.Self;
 import io.druid.guice.annotations.Smile;
+import io.druid.java.util.common.RE;
 import io.druid.java.util.common.lifecycle.Lifecycle;
 import io.druid.java.util.common.logger.Logger;
 import io.druid.server.DruidNode;
@@ -79,6 +77,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -183,9 +182,24 @@ static Server makeJettyServer(
       Binding<SslContextFactory> sslContextFactoryBinding
   )
   {
-    final QueuedThreadPool threadPool = new QueuedThreadPool();
-    threadPool.setMinThreads(config.getNumThreads());
-    threadPool.setMaxThreads(config.getNumThreads());
+    // adjusting to make config.getNumThreads() mean, "number of threads
+    // that concurrently handle the requests".
+    int numServerThreads = config.getNumThreads() + getMaxJettyAcceptorsSelectorsNum(node);
+
+    final QueuedThreadPool threadPool;
+    if (config.getQueueSize() == Integer.MAX_VALUE) {
+      threadPool = new QueuedThreadPool();
+      threadPool.setMinThreads(numServerThreads);
+      threadPool.setMaxThreads(numServerThreads);
+    } else {
+      threadPool = new QueuedThreadPool(
+          numServerThreads,
+          numServerThreads,
+          60000, // same default is used in other case when threadPool = new QueuedThreadPool()
+          new LinkedBlockingQueue<>(config.getQueueSize())
+      );
+    }
+
     threadPool.setDaemon(true);
 
     final Server server = new Server(threadPool);
@@ -261,8 +275,8 @@ static void initializeServer(Injector injector, Lifecycle lifecycle, final Serve
     try {
       initializer.initialize(server, injector);
     }
-    catch (ConfigurationException e) {
-      throw new ProvisionException(Iterables.getFirst(e.getErrorMessages(), null).getMessage());
+    catch (Exception e) {
+      throw new RE(e, "server initialization exception");
     }
 
     lifecycle.addHandler(
@@ -288,6 +302,14 @@ public void stop()
     );
   }
 
+  private static int getMaxJettyAcceptorsSelectorsNum(DruidNode druidNode)
+  {
+    // This computation is based on Jetty v9.3.19 which uses upto 8(4 acceptors and 4 selectors) threads per
+    // ServerConnector
+    int numServerConnector = (druidNode.isEnablePlaintextPort() ? 1 : 0) + (druidNode.isEnableTlsPort() ? 1 : 0);
+    return numServerConnector * 8;
+  }
+
   @Provides
   @Singleton
   public JettyMonitor getJettyMonitor(

server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java

@@ -0,0 +1,85 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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 io.druid.server.initialization.jetty;
+
+import com.google.common.base.Preconditions;
+
+import javax.servlet.Filter;
+import javax.servlet.FilterChain;
+import javax.servlet.FilterConfig;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ */
+public class LimitRequestsFilter implements Filter
+{
+  private final int maxActiveRequests;
+
+  private final AtomicInteger activeRequestsCount = new AtomicInteger();
+
+  public LimitRequestsFilter(int maxActiveRequests)
+  {
+    Preconditions.checkArgument(
+        maxActiveRequests > 0 && maxActiveRequests < Integer.MAX_VALUE,
+        "maxActiveRequests must be > 0 and < Integer.MAX_VALUE."
+    );
+    this.maxActiveRequests = maxActiveRequests;
+  }
+
+  @Override
+  public void init(FilterConfig filterConfig) throws ServletException
+  {
+  }
+
+  @Override
+  public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
+      throws IOException, ServletException
+  {
+
+    int curr = activeRequestsCount.incrementAndGet();
+    try {
+      if (curr <= maxActiveRequests) {
+        chain.doFilter(request, response);
+      } else {
+        // See https://tools.ietf.org/html/rfc6585 for status code 429 explanation.
+        ((HttpServletResponse) response).sendError(429, "Too Many Requests");
+      }
+    }
+    finally {
+      activeRequestsCount.decrementAndGet();
+    }
+  }
+
+  @Override
+  public void destroy()
+  {
+
+  }
+
+  public int getActiveRequestsCount()
+  {
+    return activeRequestsCount.get();
+  }
+}

server/src/test/java/io/druid/server/initialization/JettyQosTest.java

@@ -80,7 +80,7 @@ public void configure(Binder binder)
                 binder.bind(AuthorizerMapper.class).toInstance(AuthTestUtils.TEST_AUTHORIZER_MAPPER);
                 JettyBindings.addQosFilter(binder, "/slow/*", 2);
                 final ServerConfig serverConfig = new ObjectMapper().convertValue(
-                    ImmutableMap.of("numThreads", "10"),
+                    ImmutableMap.of("numThreads", "2"),
                     ServerConfig.class
                 );
                 binder.bind(ServerConfig.class).toInstance(serverConfig);