Introduce flag for testing CCS compatibility

modules/lang-mustache/src/internalClusterTest/java/org/elasticsearch/script/mustache/MultiSearchTemplateIT.java

@@ -8,23 +8,30 @@
 
 package org.elasticsearch.script.mustache;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.script.ScriptType;
+import org.elasticsearch.script.mustache.MultiSearchTemplateResponse.Item;
+import org.elasticsearch.search.DummyQueryParserPlugin;
+import org.elasticsearch.search.SearchService;
 import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.xcontent.json.JsonXContent;
 
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
 import static org.hamcrest.Matchers.arrayWithSize;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.instanceOf;
@@ -34,7 +41,15 @@ public class MultiSearchTemplateIT extends ESIntegTestCase {
 
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
-        return Collections.singleton(MustachePlugin.class);
+        return List.of(MustachePlugin.class, DummyQueryParserPlugin.class);
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal, otherSettings))
+            .put(SearchService.CCS_VERSION_CHECK_SETTING.getKey(), "true")
+            .build();
     }
 
     public void testBasic() throws Exception {
@@ -164,4 +179,27 @@ public void testBasic() throws Exception {
         assertThat(searchTemplateResponse5.hasResponse(), is(false));
         assertThat(searchTemplateResponse5.getSource().utf8ToString(), equalTo("{\"query\":{\"terms\":{\"group\":[1,2,3,]}}}"));
     }
+
+    /**
+    * Test that triggering the CCS compatibility check with a query that shouldn't go to the minor before Version.CURRENT works
+    */
+    public void testCCSCheckCompatibility() throws Exception {
+        String templateString = """
+            {
+            "source": "{ \\"query\\":{\\"fail_before_current_version\\":{}} }"
+            }""";
+        SearchTemplateRequest searchTemplateRequest = SearchTemplateRequest.fromXContent(
+            createParser(JsonXContent.jsonXContent, templateString)
+        );
+        searchTemplateRequest.setRequest(new SearchRequest());
+        MultiSearchTemplateRequest request = new MultiSearchTemplateRequest();
+        request.add(searchTemplateRequest);
+        MultiSearchTemplateResponse multiSearchTemplateResponse = client().execute(MultiSearchTemplateAction.INSTANCE, request).get();
+        Item response = multiSearchTemplateResponse.getResponses()[0];
+        assertTrue(response.isFailure());
+        Exception ex = response.getFailure();
+        assertThat(ex.getMessage(), containsString("[class org.elasticsearch.action.search.SearchRequest] is not compatible with version"));
+        assertThat(ex.getMessage(), containsString("'search.check_ccs_compatibility' setting is enabled."));
+        assertEquals("This query isn't serializable to nodes before " + Version.CURRENT, ex.getCause().getMessage());
+    }
 }

modules/lang-mustache/src/internalClusterTest/java/org/elasticsearch/script/mustache/SearchTemplateIT.java

@@ -8,12 +8,16 @@
 package org.elasticsearch.script.mustache;
 
 import org.elasticsearch.ResourceNotFoundException;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.admin.cluster.storedscripts.GetStoredScriptResponse;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.script.ScriptType;
+import org.elasticsearch.search.DummyQueryParserPlugin;
+import org.elasticsearch.search.SearchService;
 import org.elasticsearch.test.ESSingleNodeTestCase;
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xcontent.json.JsonXContent;
@@ -23,7 +27,9 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
@@ -38,7 +44,12 @@ public class SearchTemplateIT extends ESSingleNodeTestCase {
 
     @Override
     protected Collection<Class<? extends Plugin>> getPlugins() {
-        return Collections.singleton(MustachePlugin.class);
+        return List.of(MustachePlugin.class, DummyQueryParserPlugin.class);
+    }
+
+    @Override
+    protected Settings nodeSettings() {
+        return Settings.builder().put(SearchService.CCS_VERSION_CHECK_SETTING.getKey(), "true").build();
     }
 
     @Before
@@ -346,4 +357,26 @@ public void testIndexedTemplateWithArray() throws Exception {
         assertHitCount(searchResponse.getResponse(), 5);
     }
 
+    /**
+     * Test that triggering the CCS compatibility check with a query that shouldn't go to the minor before Version.CURRENT works
+     */
+    public void testCCSCheckCompatibility() throws Exception {
+        String templateString = """
+            {
+              "source": "{ \\"query\\":{\\"fail_before_current_version\\":{}} }"
+            }""";
+        SearchTemplateRequest request = SearchTemplateRequest.fromXContent(createParser(JsonXContent.jsonXContent, templateString));
+        request.setRequest(new SearchRequest());
+        ExecutionException ex = expectThrows(
+            ExecutionException.class,
+            () -> client().execute(SearchTemplateAction.INSTANCE, request).get()
+        );
+        assertThat(
+            ex.getCause().getMessage(),
+            containsString("[class org.elasticsearch.action.search.SearchRequest] is not compatible with version")
+        );
+        assertThat(ex.getCause().getMessage(), containsString("'search.check_ccs_compatibility' setting is enabled."));
+        assertEquals("This query isn't serializable to nodes before " + Version.CURRENT, ex.getCause().getCause().getMessage());
+    }
+
 }

server/src/internalClusterTest/java/org/elasticsearch/index/query/plugin/CustomQueryParserIT.java

@@ -10,6 +10,8 @@
 
 import org.elasticsearch.index.query.BoolQueryBuilder;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.search.DummyQueryBuilder;
+import org.elasticsearch.search.DummyQueryParserPlugin;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.junit.Before;
 

server/src/internalClusterTest/java/org/elasticsearch/search/msearch/MultiSearchIT.java

@@ -8,9 +8,13 @@
 
 package org.elasticsearch.search.msearch;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.search.MultiSearchRequest;
 import org.elasticsearch.action.search.MultiSearchResponse;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.DummyQueryBuilder;
+import org.elasticsearch.search.SearchService;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.xcontent.XContentType;
 
@@ -22,6 +26,14 @@
 
 public class MultiSearchIT extends ESIntegTestCase {
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal, otherSettings))
+            .put(SearchService.CCS_VERSION_CHECK_SETTING.getKey(), "true")
+            .build();
+    }
+
     public void testSimpleMultiSearch() {
         createIndex("test");
         ensureGreen();
@@ -70,4 +82,32 @@ public void testSimpleMultiSearchMoreRequests() {
         }
     }
 
+    /**
+     * Test that triggering the CCS compatibility check with a query that shouldn't go to the minor before Version.CURRENT works
+     */
+    public void testCCSCheckCompatibility() throws Exception {
+        createIndex("test");
+        ensureGreen();
+        client().prepareIndex("test").setId("1").setSource("field", "xxx").get();
+        client().prepareIndex("test").setId("2").setSource("field", "yyy").get();
+        refresh();
+        MultiSearchResponse response = client().prepareMultiSearch()
+            .add(client().prepareSearch("test").setQuery(QueryBuilders.termQuery("field", "xxx")))
+            .add(client().prepareSearch("test").setQuery(QueryBuilders.termQuery("field", "yyy")))
+            .add(client().prepareSearch("test").setQuery(new DummyQueryBuilder() {
+                @Override
+                public Version getMinimalSupportedVersion() {
+                    return Version.CURRENT;
+                }
+            }))
+            .get();
+
+        assertThat(response.getResponses().length, equalTo(3));
+        assertHitCount(response.getResponses()[0].getResponse(), 1L);
+        assertHitCount(response.getResponses()[1].getResponse(), 1L);
+        assertTrue(response.getResponses()[2].isFailure());
+        assertTrue(
+            response.getResponses()[2].getFailure().getMessage().contains("the 'search.check_ccs_compatibility' setting is enabled")
+        );
+    }
 }

server/src/main/java/org/elasticsearch/action/admin/indices/resolve/ResolveIndexAction.java

@@ -34,6 +34,7 @@
 import org.elasticsearch.common.util.concurrent.CountDown;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.Index;
+import org.elasticsearch.search.SearchService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.RemoteClusterAware;
@@ -54,6 +55,8 @@
 import java.util.SortedMap;
 import java.util.TreeMap;
 
+import static org.elasticsearch.action.search.TransportSearchHelper.checkCCSVersionCompatibility;
+
 public class ResolveIndexAction extends ActionType<ResolveIndexAction.Response> {
 
     public static final ResolveIndexAction INSTANCE = new ResolveIndexAction();
@@ -436,6 +439,7 @@ public static class TransportAction extends HandledTransportAction<Request, Resp
         private final ClusterService clusterService;
         private final RemoteClusterService remoteClusterService;
         private final IndexAbstractionResolver indexAbstractionResolver;
+        private final boolean ccsCheckCompatibility;
 
         @Inject
         public TransportAction(
@@ -450,10 +454,14 @@ public TransportAction(
             this.clusterService = clusterService;
             this.remoteClusterService = transportService.getRemoteClusterService();
             this.indexAbstractionResolver = new IndexAbstractionResolver(indexNameExpressionResolver);
+            this.ccsCheckCompatibility = SearchService.CCS_VERSION_CHECK_SETTING.get(clusterService.getSettings());
         }
 
         @Override
         protected void doExecute(Task task, Request request, final ActionListener<Response> listener) {
+            if (ccsCheckCompatibility) {
+                checkCCSVersionCompatibility(request);
+            }
             final ClusterState clusterState = clusterService.state();
             final Map<String, OriginalIndices> remoteClusterIndices = remoteClusterService.groupIndices(
                 request.indicesOptions(),

server/src/main/java/org/elasticsearch/action/fieldcaps/TransportFieldCapabilitiesAction.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.search.SearchService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.RemoteClusterAware;
@@ -45,6 +46,8 @@
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
+import static org.elasticsearch.action.search.TransportSearchHelper.checkCCSVersionCompatibility;
+
 public class TransportFieldCapabilitiesAction extends HandledTransportAction<FieldCapabilitiesRequest, FieldCapabilitiesResponse> {
     public static final String ACTION_NODE_NAME = FieldCapabilitiesAction.NAME + "[n]";
 
@@ -55,6 +58,7 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
 
     private final FieldCapabilitiesFetcher fieldCapabilitiesFetcher;
     private final Predicate<String> metadataFieldPred;
+    private final boolean ccsCheckCompatibility;
 
     @Inject
     public TransportFieldCapabilitiesAction(
@@ -79,10 +83,14 @@ public TransportFieldCapabilitiesAction(
             FieldCapabilitiesNodeRequest::new,
             new NodeTransportHandler()
         );
+        this.ccsCheckCompatibility = SearchService.CCS_VERSION_CHECK_SETTING.get(clusterService.getSettings());
     }
 
     @Override
     protected void doExecute(Task task, FieldCapabilitiesRequest request, final ActionListener<FieldCapabilitiesResponse> listener) {
+        if (ccsCheckCompatibility) {
+            checkCCSVersionCompatibility(request);
+        }
         // retrieve the initial timestamp in case the action is a cross cluster search
         long nowInMillis = request.nowInMillis() == null ? System.currentTimeMillis() : request.nowInMillis();
         final ClusterState clusterState = clusterService.state();

server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java

@@ -95,6 +95,7 @@
 
 import static org.elasticsearch.action.search.SearchType.DFS_QUERY_THEN_FETCH;
 import static org.elasticsearch.action.search.SearchType.QUERY_THEN_FETCH;
+import static org.elasticsearch.action.search.TransportSearchHelper.checkCCSVersionCompatibility;
 import static org.elasticsearch.search.sort.FieldSortBuilder.hasPrimaryFieldSort;
 import static org.elasticsearch.threadpool.ThreadPool.Names.SYSTEM_CRITICAL_READ;
 import static org.elasticsearch.threadpool.ThreadPool.Names.SYSTEM_READ;
@@ -132,6 +133,7 @@ public class TransportSearchAction extends HandledTransportAction<SearchRequest,
     private final CircuitBreaker circuitBreaker;
     private final ExecutorSelector executorSelector;
     private final int defaultPreFilterShardSize;
+    private final boolean ccsCheckCompatibility;
 
     @Inject
     public TransportSearchAction(
@@ -160,6 +162,7 @@ public TransportSearchAction(
         this.namedWriteableRegistry = namedWriteableRegistry;
         this.executorSelector = executorSelector;
         this.defaultPreFilterShardSize = DEFAULT_PRE_FILTER_SHARD_SIZE.get(clusterService.getSettings());
+        this.ccsCheckCompatibility = SearchService.CCS_VERSION_CHECK_SETTING.get(clusterService.getSettings());
     }
 
     private Map<String, OriginalIndices> buildPerIndexOriginalIndices(
@@ -371,6 +374,9 @@ private void executeRequest(
         ActionListener<SearchRequest> rewriteListener = ActionListener.wrap(rewritten -> {
             final SearchContextId searchContext;
             final Map<String, OriginalIndices> remoteClusterIndices;
+            if (ccsCheckCompatibility) {
+                checkCCSVersionCompatibility(rewritten);
+            }
             if (rewritten.pointInTimeBuilder() != null) {
                 searchContext = rewritten.pointInTimeBuilder().getSearchContextId(namedWriteableRegistry);
                 remoteClusterIndices = getIndicesFromSearchContexts(searchContext, rewritten.indicesOptions());