query laning and load shedding

benchmarks/src/test/java/org/apache/druid/benchmark/query/CachingClusteredClientBenchmark.java

@@ -104,7 +104,10 @@
 import org.apache.druid.query.topn.TopNResultValue;
 import org.apache.druid.segment.QueryableIndex;
 import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.server.QueryScheduler;
 import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.initialization.ServerConfig;
+import org.apache.druid.server.scheduling.NoQueryLaningStrategy;
 import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.timeline.DataSegment.PruneSpecsHolder;
 import org.apache.druid.timeline.SegmentId;
@@ -338,7 +341,8 @@ public <T, QueryType extends Query<T>> QueryToolChest<T, QueryType> getToolChest
         new CacheConfig(),
         new DruidHttpClientConfig(),
         processingConfig,
-        forkJoinPool
+        forkJoinPool,
+        new QueryScheduler(0, NoQueryLaningStrategy.INSTANCE, new ServerConfig())
     );
   }
 

distribution/bin/check-licenses.py

@@ -214,6 +214,7 @@ def build_compatible_license_names():
     compatible_licenses['Apache License, Version 2.0'] = 'Apache License version 2.0'
     compatible_licenses['The Apache Software License, Version 2.0'] = 'Apache License version 2.0'
     compatible_licenses['Apache 2.0'] = 'Apache License version 2.0'
+    compatible_licenses['Apache-2.0'] = 'Apache License version 2.0'
     compatible_licenses['Apache 2'] = 'Apache License version 2.0'
     compatible_licenses['Apache License 2.0'] = 'Apache License version 2.0'
     compatible_licenses['Apache Software License - Version 2.0'] = 'Apache License version 2.0'

docs/configuration/index.md

@@ -1481,9 +1481,35 @@ These Broker configurations can be defined in the `broker/runtime.properties` fi
 |`druid.broker.select.tier`|`highestPriority`, `lowestPriority`, `custom`|If segments are cross-replicated across tiers in a cluster, you can tell the broker to prefer to select segments in a tier with a certain priority.|`highestPriority`|
 |`druid.broker.select.tier.custom.priorities`|`An array of integer priorities.`|Select servers in tiers with a custom priority list.|None|
 
+##### Query laning
+
+*Laning strategies* allow you to control capacity utilization for heterogeneous query workloads. With laning, the broker examines and classifies a query for the purpose of assigning it to a 'lane'. Lanes have capacity limits, enforced by the broker, that can be used to ensure sufficient resources are available for other lanes or for interactive queries (with no lane), or to limit overall throughput for queries within the lane. Requests in excess of the capacity are discarded with an HTTP 429 status code.
+
+|Property|Description|Default|
+|--------|-----------|-------|
+|`druid.query.scheduler.numThreads`|Maximum number of HTTP threads to dedicate to query processing. To save HTTP thread capacity, this should be lower than `druid.server.http.numThreads`.|Unbounded|
+|`druid.query.scheduler.laning.strategy`|Query laning strategy to use to assign queries to a lane in order to control capacities for certain classes of queries.|`none`|
+
+##### Laning strategies
+
+###### No laning strategy
+
+In this mode, queries are never assigned a lane, and the concurrent query count will only be limited by `druid.server.http.numThreads` or `druid.query.scheduler.numThreads`, if set. This is the default Druid query scheduler operating mode. Enable this strategy explicitly by setting `druid.query.scheduler.laning.strategy` to `none`.
+
+###### 'High/Low' laning strategy
+This laning strategy splits queries with a `priority` below zero into a `low` query lane, automatically. Queries with priority of zero (the default) or above are considered 'interactive'. The limit on `low` queries can be set to some desired percentage of the total capacity (or HTTP thread pool size), reserving capacity for interactive queries. Queries in the `low` lane are _not_ guaranteed their capacity, which may be consumed by interactive queries, but may use up to this limit if total capacity is available.
+
+If the `low` lane is specified in the [query context](../querying/query-context.md) `lane` parameter, this will override the computed lane. 
+
+This strategy can be enabled by setting `druid.query.scheduler.laning.strategy=hilo`.
+
+|Property|Description|Default|
+|--------|-----------|-------|
+|`druid.query.scheduler.laning.maxLowPercent`|Maximum percent of the smaller number of`druid.server.http.numThreads` or `druid.query.scheduler.numThreads`, defining the number of HTTP threads that can be used by queries with a priority lower than 0. Value must be in the range 1 to 100, and will be rounded up|No default, must be set if using this mode|
+
 ##### Server Configuration
 
-Druid uses Jetty to serve HTTP requests.
+Druid uses Jetty to serve HTTP requests. Each query being processed consumes a single thread from `druid.server.http.numThreads`, so consider defining `druid.query.scheduler.numThreads` to a lower value in order to reserve HTTP threads for responding to health checks, lookup loading, and other non-query, and in most cases comparatively very short lived, HTTP requests.
 
 |Property|Description|Default|
 |--------|-----------|-------|

docs/querying/query-context.md

@@ -29,6 +29,7 @@ The query context is used for various query configuration parameters. The follow
 |-----------------|----------------------------------------|----------------------|
 |timeout          | `druid.server.http.defaultQueryTimeout`| Query timeout in millis, beyond which unfinished queries will be cancelled. 0 timeout means `no timeout`. To set the default timeout, see [Broker configuration](../configuration/index.html#broker) |
 |priority         | `0`                                    | Query Priority. Queries with higher priority get precedence for computational resources.|
+|lane             | `null`                                 | Query lane, used to control usage limits on classes of queries. See [Broker configuration](../configuration/index.html#broker) for more details.|
 |queryId          | auto-generated                         | Unique identifier given to this query. If a query ID is set or known, this can be used to cancel the query |
 |useCache         | `true`                                 | Flag indicating whether to leverage the query cache for this query. When set to false, it disables reading from the query cache for this query. When set to true, Apache Druid uses `druid.broker.cache.useCache` or `druid.historical.cache.useCache` to determine whether or not to read from the query cache |
 |populateCache    | `true`                                 | Flag indicating whether to save the results of the query to the query cache. Primarily used for debugging. When set to false, it disables saving the results of this query to the query cache. When set to true, Druid uses `druid.broker.cache.populateCache` or `druid.historical.cache.populateCache` to determine whether or not to save the results of this query to the query cache |

docs/querying/querying.md

@@ -109,6 +109,8 @@ If a query fails, you will get an HTTP 500 response containing a JSON object wit
 }
 ```
 
+If a query request fails due to being limited by the [query scheduler laning configuration](../configuration/index.md#broker), an HTTP 429 response with the same JSON object schema as an error response, but with `errorMessage` of the form: "Total query capacity exceeded" or "Query capacity exceeded for lane 'low'".
+
 The fields in the response are:
 
 |field|description|

extensions-contrib/moving-average-query/src/test/java/org/apache/druid/query/movingaverage/MovingAverageQueryTest.java

@@ -65,7 +65,9 @@
 import org.apache.druid.query.timeseries.TimeseriesQuery;
 import org.apache.druid.query.timeseries.TimeseriesResultValue;
 import org.apache.druid.server.ClientQuerySegmentWalker;
+import org.apache.druid.server.QueryScheduler;
 import org.apache.druid.server.initialization.ServerConfig;
+import org.apache.druid.server.scheduling.NoQueryLaningStrategy;
 import org.apache.druid.testing.InitializedNullHandlingTest;
 import org.apache.druid.timeline.TimelineLookup;
 import org.hamcrest.core.IsInstanceOf;
@@ -361,7 +363,8 @@ public String getFormatString()
             return null;
           }
         },
-        ForkJoinPool.commonPool()
+        ForkJoinPool.commonPool(),
+        new QueryScheduler(0, NoQueryLaningStrategy.INSTANCE, new ServerConfig())
     );
 
     ClientQuerySegmentWalker walker = new ClientQuerySegmentWalker(

indexing-service/src/test/java/org/apache/druid/indexing/common/task/RealtimeIndexTaskTest.java

@@ -909,7 +909,7 @@ private TaskToolbox makeToolbox(
                 new QueryWatcher()
                 {
                   @Override
-                  public void registerQuery(Query query, ListenableFuture future)
+                  public void registerQueryFuture(Query query, ListenableFuture future)
                   {
                     // do nothing
                   }

licenses.yaml

@@ -1869,6 +1869,17 @@ libraries:
 
 ---
 
+name: Resilience4j
+license_category: binary
+module: java-core
+license_name: Apache License version 2.0
+version: 1.3.1
+libraries:
+  - io.github.resilience4j: resilience4j-core
+  - io.github.resilience4j: resilience4j-bulkhead
+
+---
+
 name: RoaringBitmap
 license_category: binary
 module: java-core
@@ -1880,6 +1891,17 @@ libraries:
 
 ---
 
+name: vavr
+license_category: binary
+module: java-core
+license_name: Apache License version 2.0
+version: 0.10.2
+libraries:
+  - io.vavr: vavr
+  - io.vavr: vavr-match
+
+---
+
 name: Config Magic
 license_category: binary
 module: java-core

pom.xml

@@ -94,6 +94,7 @@
         <codehaus.jackson.version>1.9.13</codehaus.jackson.version>
         <log4j.version>2.8.2</log4j.version>
         <netty3.version>3.10.6.Final</netty3.version>
+        <resilience4j.version>1.3.1</resilience4j.version>
         <!-- Spark updated in https://github.com/apache/spark/pull/19884 -->
         <netty4.version>4.1.45.Final</netty4.version>
         <node.version>v10.14.2</node.version>
@@ -1181,6 +1182,11 @@
                     </exclusion>
                 </exclusions>
             </dependency>
+            <dependency>
+                <groupId>io.github.resilience4j</groupId>
+                <artifactId>resilience4j-bulkhead</artifactId>
+                <version>${resilience4j.version}</version>
+            </dependency>
 
             <dependency>
                 <groupId>org.testng</groupId>

processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java

@@ -144,7 +144,7 @@ public Iterable<T> call()
                 )
             );
 
-            queryWatcher.registerQuery(query, futures);
+            queryWatcher.registerQueryFuture(query, futures);
 
             try {
               return new MergeIterable<>(

processing/src/main/java/org/apache/druid/query/GroupByMergedQueryRunner.java

@@ -178,7 +178,7 @@ private void waitForFutureCompletion(
   )
   {
     try {
-      queryWatcher.registerQuery(query, future);
+      queryWatcher.registerQueryFuture(query, future);
       if (QueryContexts.hasTimeout(query)) {
         future.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS);
       } else {

processing/src/main/java/org/apache/druid/query/Query.java

@@ -21,6 +21,7 @@
 
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Ordering;
 import org.apache.druid.guice.annotations.ExtensionPoint;
 import org.apache.druid.java.util.common.granularity.Granularity;
@@ -146,4 +147,10 @@ default Query<T> optimizeForSegment(PerSegmentQueryOptimizationContext optimizat
   {
     return this;
   }
+
+  default Query<T> withLane(String lane)
+  {
+    return withOverriddenContext(ImmutableMap.of(QueryContexts.LANE_KEY, lane));
+  }
+
 }

processing/src/main/java/org/apache/druid/query/QueryContexts.java

@@ -35,6 +35,7 @@
 public class QueryContexts
 {
   public static final String PRIORITY_KEY = "priority";
+  public static final String LANE_KEY = "lane";
   public static final String TIMEOUT_KEY = "timeout";
   public static final String MAX_SCATTER_GATHER_BYTES_KEY = "maxScatterGatherBytes";
   public static final String MAX_QUEUED_BYTES_KEY = "maxQueuedBytes";
@@ -202,6 +203,11 @@ public static <T> int getPriority(Query<T> query, int defaultValue)
     return parseInt(query, PRIORITY_KEY, defaultValue);
   }
 
+  public static <T> String getLane(Query<T> query)
+  {
+    return (String) query.getContextValue(LANE_KEY);
+  }
+
   public static <T> boolean getEnableParallelMerges(Query<T> query)
   {
     return parseBoolean(query, BROKER_PARALLEL_MERGE_KEY, DEFAULT_ENABLE_PARALLEL_MERGE);

processing/src/main/java/org/apache/druid/query/QueryException.java

@@ -0,0 +1,85 @@
+/*
+ * 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.query;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+/**
+ * Base serializable error response
+ *
+ * QueryResource and SqlResource are expected to emit the JSON form of this object when errors happen.
+ */
+public class QueryException extends RuntimeException
+{
+  private final String errorCode;
+  private final String errorClass;
+  private final String host;
+
+  public QueryException(Throwable cause, String errorCode, String errorClass, String host)
+  {
+    super(cause == null ? null : cause.getMessage(), cause);
+    this.errorCode = errorCode;
+    this.errorClass = errorClass;
+    this.host = host;
+  }
+
+  @JsonCreator
+  public QueryException(
+      @JsonProperty("error") @Nullable String errorCode,
+      @JsonProperty("errorMessage") String errorMessage,
+      @JsonProperty("errorClass") @Nullable String errorClass,
+      @JsonProperty("host") @Nullable String host
+  )
+  {
+    super(errorMessage);
+    this.errorCode = errorCode;
+    this.errorClass = errorClass;
+    this.host = host;
+  }
+
+  @Nullable
+  @JsonProperty("error")
+  public String getErrorCode()
+  {
+    return errorCode;
+  }
+
+  @JsonProperty("errorMessage")
+  @Override
+  public String getMessage()
+  {
+    return super.getMessage();
+  }
+
+  @JsonProperty
+  public String getErrorClass()
+  {
+    return errorClass;
+  }
+
+  @JsonProperty
+  public String getHost()
+  {
+    return host;
+  }
+}

processing/src/main/java/org/apache/druid/query/QueryInterruptedException.java

@@ -42,7 +42,7 @@
  * The QueryResource is expected to emit the JSON form of this object when errors happen, and the DirectDruidClient
  * deserializes and wraps them.
  */
-public class QueryInterruptedException extends RuntimeException
+public class QueryInterruptedException extends QueryException
 {
   public static final String QUERY_INTERRUPTED = "Query interrupted";
   public static final String QUERY_TIMEOUT = "Query timeout";
@@ -52,10 +52,6 @@ public class QueryInterruptedException extends RuntimeException
   public static final String UNSUPPORTED_OPERATION = "Unsupported operation";
   public static final String UNKNOWN_EXCEPTION = "Unknown exception";
 
-  private final String errorCode;
-  private final String errorClass;
-  private final String host;
-
   @JsonCreator
   public QueryInterruptedException(
       @JsonProperty("error") @Nullable String errorCode,
@@ -64,10 +60,7 @@ public QueryInterruptedException(
       @JsonProperty("host") @Nullable String host
   )
   {
-    super(errorMessage);
-    this.errorCode = errorCode;
-    this.errorClass = errorClass;
-    this.host = host;
+    super(errorCode, errorMessage, errorClass, host);
   }
 
   /**
@@ -83,36 +76,7 @@ public QueryInterruptedException(Throwable cause)
 
   public QueryInterruptedException(Throwable cause, String host)
   {
-    super(cause == null ? null : cause.getMessage(), cause);
-    this.errorCode = getErrorCodeFromThrowable(cause);
-    this.errorClass = getErrorClassFromThrowable(cause);
-    this.host = host;
-  }
-
-  @Nullable
-  @JsonProperty("error")
-  public String getErrorCode()
-  {
-    return errorCode;
-  }
-
-  @JsonProperty("errorMessage")
-  @Override
-  public String getMessage()
-  {
-    return super.getMessage();
-  }
-
-  @JsonProperty
-  public String getErrorClass()
-  {
-    return errorClass;
-  }
-
-  @JsonProperty
-  public String getHost()
-  {
-    return host;
+    super(cause, getErrorCodeFromThrowable(cause), getErrorClassFromThrowable(cause), host);
   }
 
   @Override
@@ -121,9 +85,9 @@ public String toString()
     return StringUtils.format(
         "QueryInterruptedException{msg=%s, code=%s, class=%s, host=%s}",
         getMessage(),
-        errorCode,
-        errorClass,
-        host
+        getErrorCode(),
+        getErrorClass(),
+        getHost()
     );
   }