Removes ExpandCollapseSearchResponseListener, search response listeners and blocking calls

This changes removes the SearchResponseListener that was used by the ExpandCollapseSearchResponseListener to expand collapsed hits.
The removal of SearchResponseListener is not a breaking change because it was never released.
This change also replace the blocking call in ExpandCollapseSearchResponseListener by a single asynchronous multi search request. The parallelism of the expand request can be set via CollapseBuilder#max_concurrent_group_searches

Closes #23048

2 files changed, 31 insertions, 119 deletions

diff --git a/core/src/main/java/org/elasticsearch/search/collapse/CollapseBuilder.java b/core/src/main/java/org/elasticsearch/search/collapse/CollapseBuilder.java
index 75a7d6dadf..542ae2c3ab 100644
--- a/core/src/main/java/org/elasticsearch/search/collapse/CollapseBuilder.java
+++ b/core/src/main/java/org/elasticsearch/search/collapse/CollapseBuilder.java
@@ -45,17 +45,20 @@ import java.util.Objects;
 public class CollapseBuilder extends ToXContentToBytes implements Writeable {
     public static final ParseField FIELD_FIELD = new ParseField("field");
     public static final ParseField INNER_HITS_FIELD = new ParseField("inner_hits");
+    public static final ParseField MAX_CONCURRENT_GROUP_REQUESTS_FIELD = new ParseField("max_concurrent_group_searches");
     private static final ObjectParser<CollapseBuilder, QueryParseContext> PARSER =
         new ObjectParser<>("collapse", CollapseBuilder::new);
 
     static {
         PARSER.declareString(CollapseBuilder::setField, FIELD_FIELD);
+        PARSER.declareInt(CollapseBuilder::setMaxConcurrentGroupRequests, MAX_CONCURRENT_GROUP_REQUESTS_FIELD);
         PARSER.declareObject(CollapseBuilder::setInnerHits,
             (p, c) -> InnerHitBuilder.fromXContent(c), INNER_HITS_FIELD);
     }
 
     private String field;
     private InnerHitBuilder innerHit;
+    private int maxConcurrentGroupRequests = 0;
 
     private CollapseBuilder() {}
 
@@ -70,12 +73,14 @@ public class CollapseBuilder extends ToXContentToBytes implements Writeable {
 
     public CollapseBuilder(StreamInput in) throws IOException {
         this.field = in.readString();
+        this.maxConcurrentGroupRequests = in.readVInt();
         this.innerHit = in.readOptionalWriteable(InnerHitBuilder::new);
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(field);
+        out.writeVInt(maxConcurrentGroupRequests);
         out.writeOptionalWriteable(innerHit);
     }
 
@@ -84,6 +89,7 @@ public class CollapseBuilder extends ToXContentToBytes implements Writeable {
         return builder;
     }
 
+    // for object parser only
     private CollapseBuilder setField(String field) {
         if (Strings.isEmpty(field)) {
             throw new IllegalArgumentException("field name is null or empty");
@@ -97,6 +103,14 @@ public class CollapseBuilder extends ToXContentToBytes implements Writeable {
         return this;
     }
 
+    public CollapseBuilder setMaxConcurrentGroupRequests(int num) {
+        if (num < 1) {
+            throw new IllegalArgumentException("maxConcurrentGroupRequests` must be positive");
+        }
+        this.maxConcurrentGroupRequests = num;
+        return this;
+    }
+
     /**
      * The name of the field to collapse against
      */
@@ -111,6 +125,13 @@ public class CollapseBuilder extends ToXContentToBytes implements Writeable {
         return this.innerHit;
     }
 
+    /**
+     * Returns the amount of group requests that are allowed to be ran concurrently in the inner_hits phase.
+     */
+    public int getMaxConcurrentGroupRequests() {
+        return maxConcurrentGroupRequests;
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
         builder.startObject();
@@ -121,6 +142,9 @@ public class CollapseBuilder extends ToXContentToBytes implements Writeable {
 
     private void innerToXContent(XContentBuilder builder) throws IOException {
         builder.field(FIELD_FIELD.getPreferredName(), field);
+        if (maxConcurrentGroupRequests > 0) {
+            builder.field(MAX_CONCURRENT_GROUP_REQUESTS_FIELD.getPreferredName(), maxConcurrentGroupRequests);
+        }
         if (innerHit != null) {
             builder.field(INNER_HITS_FIELD.getPreferredName(), innerHit);
         }
@@ -133,13 +157,18 @@ public class CollapseBuilder extends ToXContentToBytes implements Writeable {
 
         CollapseBuilder that = (CollapseBuilder) o;
 
-        if (field != null ? !field.equals(that.field) : that.field != null) return false;
+        if (maxConcurrentGroupRequests != that.maxConcurrentGroupRequests) return false;
+        if (!field.equals(that.field)) return false;
         return innerHit != null ? innerHit.equals(that.innerHit) : that.innerHit == null;
+
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(this.field, this.innerHit);
+        int result = field.hashCode();
+        result = 31 * result + (innerHit != null ? innerHit.hashCode() : 0);
+        result = 31 * result + maxConcurrentGroupRequests;
+        return result;
     }
 
     public CollapseContext build(SearchContext context) {
diff --git a/core/src/main/java/org/elasticsearch/search/collapse/ExpandCollapseSearchResponseListener.java b/core/src/main/java/org/elasticsearch/search/collapse/ExpandCollapseSearchResponseListener.java
deleted file mode 100644
index b9caa5216c..0000000000
--- a/core/src/main/java/org/elasticsearch/search/collapse/ExpandCollapseSearchResponseListener.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * Licensed to Elasticsearch under one or more contributor
- * license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright
- * ownership. Elasticsearch 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.elasticsearch.search.collapse;
-
-import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.client.Client;
-import org.elasticsearch.index.query.BoolQueryBuilder;
-import org.elasticsearch.index.query.InnerHitBuilder;
-import org.elasticsearch.index.query.QueryBuilder;
-import org.elasticsearch.index.query.QueryBuilders;
-import org.elasticsearch.search.SearchHit;
-import org.elasticsearch.search.SearchHits;
-import org.elasticsearch.search.builder.SearchSourceBuilder;
-
-import java.util.HashMap;
-import java.util.Objects;
-import java.util.function.BiConsumer;
-
-/**
- * A search response listener that intercepts the search response and expands collapsed hits
- * using the {@link CollapseBuilder#innerHit} options.
- */
-public class ExpandCollapseSearchResponseListener implements BiConsumer<SearchRequest, SearchResponse> {
-    private final Client client;
-
-    public ExpandCollapseSearchResponseListener(Client client) {
-        this.client = Objects.requireNonNull(client);
-    }
-
-    @Override
-    public void accept(SearchRequest searchRequest, SearchResponse searchResponse) {
-        if (searchRequest.source() == null) {
-            return ;
-        }
-        CollapseBuilder collapseBuilder = searchRequest.source().collapse();
-        if (collapseBuilder == null || collapseBuilder.getInnerHit() == null) {
-            return ;
-        }
-        for (SearchHit hit : searchResponse.getHits()) {
-            BoolQueryBuilder groupQuery = new BoolQueryBuilder();
-            Object collapseValue = hit.field(collapseBuilder.getField()).getValue();
-            if (collapseValue != null) {
-                groupQuery.filter(QueryBuilders.matchQuery(collapseBuilder.getField(), collapseValue));
-            } else {
-                groupQuery.mustNot(QueryBuilders.existsQuery(collapseBuilder.getField()));
-            }
-            QueryBuilder origQuery = searchRequest.source().query();
-            if (origQuery != null) {
-                groupQuery.must(origQuery);
-            }
-            SearchSourceBuilder sourceBuilder = createGroupSearchBuilder(collapseBuilder.getInnerHit())
-                .query(groupQuery);
-            SearchRequest groupRequest = new SearchRequest(searchRequest.indices())
-                .types(searchRequest.types())
-                .source(sourceBuilder);
-            SearchResponse groupResponse = client.search(groupRequest).actionGet();
-            SearchHits innerHits = groupResponse.getHits();
-            if (hit.getInnerHits() == null) {
-                hit.setInnerHits(new HashMap<>(1));
-            }
-            hit.getInnerHits().put(collapseBuilder.getInnerHit().getName(), innerHits);
-        }
-    }
-
-    private SearchSourceBuilder createGroupSearchBuilder(InnerHitBuilder options) {
-        SearchSourceBuilder groupSource = new SearchSourceBuilder();
-        groupSource.from(options.getFrom());
-        groupSource.size(options.getSize());
-        if (options.getSorts() != null) {
-            options.getSorts().forEach(groupSource::sort);
-        }
-        if (options.getFetchSourceContext() != null) {
-            if (options.getFetchSourceContext().includes() == null && options.getFetchSourceContext().excludes() == null) {
-                groupSource.fetchSource(options.getFetchSourceContext().fetchSource());
-            } else {
-                groupSource.fetchSource(options.getFetchSourceContext().includes(),
-                    options.getFetchSourceContext().excludes());
-            }
-        }
-        if (options.getDocValueFields() != null) {
-            options.getDocValueFields().forEach(groupSource::docValueField);
-        }
-        if (options.getStoredFieldsContext() != null && options.getStoredFieldsContext().fieldNames() != null) {
-            options.getStoredFieldsContext().fieldNames().forEach(groupSource::storedField);
-        }
-        if (options.getScriptFields() != null) {
-            for (SearchSourceBuilder.ScriptField field : options.getScriptFields()) {
-                groupSource.scriptField(field.fieldName(), field.script());
-            }
-        }
-        if (options.getHighlightBuilder() != null) {
-            groupSource.highlighter(options.getHighlightBuilder());
-        }
-        groupSource.explain(options.isExplain());
-        groupSource.trackScores(options.isTrackScores());
-        return groupSource;
-    }
-
-}