[Tests] Enhance multi cluster search test skipping

qa/multi-cluster-search/build.gradle

@@ -64,6 +64,7 @@ BuildParams.bwcVersions.withWireCompatible(ccsSupportedVersion) { bwcVersion, ba
     doFirst {
       nonInputProperties.systemProperty('tests.rest.suite', 'multi_cluster')
       nonInputProperties.systemProperty('tests.rest.cluster', localCluster.map(c -> c.allHttpSocketURI.join(",")))
+      nonInputProperties.systemProperty('tests.rest.remote_cluster_version', bwcVersion.toString())
     }
     dependsOn "${baseName}#remote-cluster"
   }

qa/multi-cluster-search/src/test/java/org/elasticsearch/search/MultiClusterSearchYamlTestSuiteIT.java

@@ -13,12 +13,49 @@
 import com.carrotsearch.randomizedtesting.annotations.TimeoutSuite;
 
 import org.apache.lucene.tests.util.TimeUnits;
+import org.elasticsearch.Version;
 import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
+import org.elasticsearch.test.rest.yaml.ClientYamlTestClient;
+import org.elasticsearch.test.rest.yaml.ClientYamlTestExecutionContext;
 import org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase;
+import org.junit.BeforeClass;
 
 @TimeoutSuite(millis = 5 * TimeUnits.MINUTE) // to account for slow as hell VMs
 public class MultiClusterSearchYamlTestSuiteIT extends ESClientYamlSuiteTestCase {
 
+    private static Version remoteEsVersion = null;
+
+    @BeforeClass
+    public static void determineRemoteClusterMinimumVersion() {
+        String remoteClusterVersion = System.getProperty("tests.rest.remote_cluster_version");
+        if (remoteClusterVersion != null) {
+            remoteEsVersion = Version.fromString(remoteClusterVersion);
+        }
+    }
+
+    protected ClientYamlTestExecutionContext createRestTestExecutionContext(
+        ClientYamlTestCandidate clientYamlTestCandidate,
+        ClientYamlTestClient clientYamlTestClient
+    ) {
+        return new ClientYamlTestExecutionContext(clientYamlTestCandidate, clientYamlTestClient, randomizeContentType()) {
+
+            /**
+             * Since the esVersion is used to skip tests in ESClientYamlSuiteTestCase, we also take into account the
+             * remote cluster version here and return it if it is lower than the local client version. This is used to
+             * skip tests if some feature isn't available on the remote cluster yet.
+             */
+            @Override
+            public Version esVersion() {
+                Version clientEsVersion = clientYamlTestClient.getEsVersion();
+                if (remoteEsVersion == null) {
+                    return clientEsVersion;
+                } else {
+                    return remoteEsVersion.before(clientEsVersion) ? remoteEsVersion : clientEsVersion;
+                }
+            }
+        };
+    }
+
     @Override
     protected boolean preserveIndicesUponCompletion() {
         return true;

test/yaml-rest-runner/src/main/java/org/elasticsearch/test/rest/yaml/ClientYamlTestExecutionContext.java

@@ -50,7 +50,7 @@ public class ClientYamlTestExecutionContext {
 
     private final boolean randomizeContentType;
 
-    ClientYamlTestExecutionContext(
+    public ClientYamlTestExecutionContext(
         ClientYamlTestCandidate clientYamlTestCandidate,
         ClientYamlTestClient clientYamlTestClient,
         boolean randomizeContentType

test/yaml-rest-runner/src/main/java/org/elasticsearch/test/rest/yaml/ESClientYamlSuiteTestCase.java

@@ -140,7 +140,7 @@ public void initAndResetContext() throws Exception {
                 os
             );
             clientYamlTestClient = initClientYamlTestClient(restSpec, client(), hosts, esVersion, masterVersion, os);
-            restTestExecutionContext = new ClientYamlTestExecutionContext(testCandidate, clientYamlTestClient, randomizeContentType());
+            restTestExecutionContext = createRestTestExecutionContext(testCandidate, clientYamlTestClient);
             adminExecutionContext = new ClientYamlTestExecutionContext(testCandidate, clientYamlTestClient, false);
             final String[] blacklist = resolvePathsProperty(REST_TESTS_BLACKLIST, null);
             blacklistPathMatchers = new ArrayList<>();
@@ -162,6 +162,16 @@ public void initAndResetContext() throws Exception {
         restTestExecutionContext.clear();
     }
 
+    /**
+     * Create the test execution context. Can be overwritten in sub-implementations of the test if the context needs to be modified.
+     */
+    protected ClientYamlTestExecutionContext createRestTestExecutionContext(
+        ClientYamlTestCandidate clientYamlTestCandidate,
+        ClientYamlTestClient clientYamlTestClient
+    ) {
+        return new ClientYamlTestExecutionContext(clientYamlTestCandidate, clientYamlTestClient, randomizeContentType());
+    }
+
     protected ClientYamlTestClient initClientYamlTestClient(
         final ClientYamlSuiteRestSpec restSpec,
         final RestClient restClient,