Added soft limit to open scroll contexts #25244

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -387,6 +387,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                     SearchService.MAX_KEEPALIVE_SETTING,
                     MultiBucketConsumerService.MAX_BUCKET_SETTING,
                     SearchService.LOW_LEVEL_CANCELLATION_SETTING,
+                    SearchService.MAX_OPEN_SCROLL_CONTEXT,
                     Node.WRITE_PORTS_FILE_SETTING,
                     Node.NODE_NAME_SETTING,
                     Node.NODE_DATA_SETTING,

server/src/main/java/org/elasticsearch/search/SearchService.java

@@ -112,6 +112,7 @@
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
@@ -145,6 +146,9 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
     public static final Setting<Boolean> DEFAULT_ALLOW_PARTIAL_SEARCH_RESULTS =
             Setting.boolSetting("search.default_allow_partial_results", true, Property.Dynamic, Property.NodeScope);
 
+    public static final Setting<Integer> MAX_OPEN_SCROLL_CONTEXT =
+        Setting.intSetting("search.max_open_scroll_context", 150, 0, Property.NodeScope);
+
 
     private final ThreadPool threadPool;
 
@@ -182,6 +186,8 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
 
     private final MultiBucketConsumerService multiBucketConsumerService;
 
+    private final AtomicInteger openScrollContexts = new AtomicInteger();
+
     public SearchService(ClusterService clusterService, IndicesService indicesService,
                          ThreadPool threadPool, ScriptService scriptService, BigArrays bigArrays, FetchPhase fetchPhase,
                          ResponseCollectorService responseCollectorService) {
@@ -592,11 +598,19 @@ private SearchContext findContext(long id, TransportRequest request) throws Sear
     }
 
     final SearchContext createAndPutContext(ShardSearchRequest request) throws IOException {
+        if (request.scroll() != null && openScrollContexts.get() >= MAX_OPEN_SCROLL_CONTEXT.get(clusterService.getSettings())) {
+            throw new ElasticsearchException(
+                "Trying to create too many scroll contexts. Must be less than or equal to: [" +
+                    MAX_OPEN_SCROLL_CONTEXT.get(clusterService.getSettings()) + "]. " + "This limit can be set by changing the ["
+                    + MAX_OPEN_SCROLL_CONTEXT.getKey() + "] setting.");
+        }
+
         SearchContext context = createContext(request);
         boolean success = false;
         try {
             putContext(context);
             if (request.scroll() != null) {
+                openScrollContexts.incrementAndGet();
                 context.indexShard().getSearchOperationListener().onNewScrollContext(context);
             }
             context.indexShard().getSearchOperationListener().onNewContext(context);
@@ -696,6 +710,7 @@ public boolean freeContext(long id) {
                 assert context.refCount() > 0 : " refCount must be > 0: " + context.refCount();
                 context.indexShard().getSearchOperationListener().onFreeContext(context);
                 if (context.scrollContext() != null) {
+                    openScrollContexts.decrementAndGet();
                     context.indexShard().getSearchOperationListener().onFreeScrollContext(context);
                 }
                 return true;

server/src/test/java/org/elasticsearch/search/SearchServiceTests.java

@@ -22,12 +22,14 @@
 
 import org.apache.lucene.search.Query;
 import org.apache.lucene.store.AlreadyClosedException;
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.search.SearchPhaseExecutionException;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.SearchTask;
 import org.elasticsearch.action.search.SearchType;
+import org.elasticsearch.action.search.ClearScrollRequest;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.WriteRequest;
@@ -75,6 +77,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.LinkedList;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
@@ -416,6 +419,44 @@ searchSourceBuilder, new String[0], false, new AliasFilter(null, Strings.EMPTY_A
         }
     }
 
+    /**
+     * test that creating more than the allowed number of scroll contexts throws an exception
+     */
+    public void testMaxOpenScrollContexts() throws RuntimeException {
+        createIndex("index");
+        client().prepareIndex("index", "type", "1").setSource("field", "value").setRefreshPolicy(IMMEDIATE).get();
+
+        final SearchService service = getInstanceFromNode(SearchService.class);
+        final IndicesService indicesService = getInstanceFromNode(IndicesService.class);
+        final IndexService indexService = indicesService.indexServiceSafe(resolveIndex("index"));
+        final IndexShard indexShard = indexService.getShard(0);
+
+        // Open all possible scrolls, clear some of them, then open more until the limit is reached
+        LinkedList<String> clearScrollIds = new LinkedList<>();
+
+        for (int i = 0; i < SearchService.MAX_OPEN_SCROLL_CONTEXT.get(Settings.EMPTY); i++) {
+            SearchResponse searchResponse = client().prepareSearch("index").setSize(1).setScroll("1m").get();
+
+            if (randomInt(4) == 0) clearScrollIds.addLast(searchResponse.getScrollId());
+        }
+
+        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
+        clearScrollRequest.setScrollIds(clearScrollIds);
+        client().clearScroll(clearScrollRequest);
+
+        for (int i = 0; i < clearScrollIds.size(); i++) {
+            client().prepareSearch("index").setSize(1).setScroll("1m").get();
+        }
+
+        ElasticsearchException ex = expectThrows(ElasticsearchException.class,
+            () -> service.createAndPutContext(new ShardScrollRequestTest(indexShard.shardId())));
+        assertEquals(
+            "Trying to create too many scroll contexts. Must be less than or equal to: [" +
+                SearchService.MAX_OPEN_SCROLL_CONTEXT.get(Settings.EMPTY) + "]. " +
+                "This limit can be set by changing the [search.max_open_scroll_context] setting.",
+            ex.getMessage());
+    }
+
     public static class FailOnRewriteQueryPlugin extends Plugin implements SearchPlugin {
         @Override
         public List<QuerySpec<?>> getQueries() {
@@ -471,6 +512,22 @@ public String getWriteableName() {
         }
     }
 
+    public static class ShardScrollRequestTest extends ShardSearchLocalRequest {
+        private Scroll scroll;
+
+        ShardScrollRequestTest(ShardId shardId) {
+            super(shardId, 1, SearchType.DEFAULT, new SearchSourceBuilder(),
+                new String[0], false, new AliasFilter(null, Strings.EMPTY_ARRAY), 1f, true, null, null);
+
+            this.scroll = new Scroll(TimeValue.timeValueMinutes(1));
+        }
+
+        @Override
+        public Scroll scroll() {
+            return this.scroll;
+        }
+    }
+
     public void testCanMatch() throws IOException {
         createIndex("index");
         final SearchService service = getInstanceFromNode(SearchService.class);