[Tests] Enhance multi cluster search test skipping (#85566)

Currently the "skip" section in out rest yaml tests take into account the lowest
minor version of nodes in the connected local cluster. For some multi cluster
CCS test we will also need the ability to skip certain tests based on the
connected remote cluster version, e.g. if in bwc tests some functionality isn't
available yet on some bwc versions we test against. This change adds that
ability to yaml rest test in the :qa:multi-cluster-search module.

Relates to #84481

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.ParametersFactory;
 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 abstract class ESClientYamlSuiteTestCase extends ESRestTestCase {
                 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 abstract class ESClientYamlSuiteTestCase extends ESRestTestCase {
         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,