Add JUnit rule based integration test cluster orchestration framework (#92379)

This commit adds a new test framework for configuring and orchestrating
test clusters for both Java and YAML REST testing. This will eventually
replace the existing "test-clusters" Gradle plugin and the build-time
cluster orchestration.

build-tools-internal/build.gradle

@@ -115,6 +115,10 @@ gradlePlugin {
       id = 'elasticsearch.java'
       implementationClass = 'org.elasticsearch.gradle.internal.ElasticsearchJavaPlugin'
     }
+    legacyInternalJavaRestTest {
+      id = 'elasticsearch.legacy-java-rest-test'
+      implementationClass = 'org.elasticsearch.gradle.internal.test.rest.LegacyJavaRestTestPlugin'
+    }
     internalJavaRestTest {
       id = 'elasticsearch.internal-java-rest-test'
       implementationClass = 'org.elasticsearch.gradle.internal.test.rest.InternalJavaRestTestPlugin'
@@ -167,9 +171,17 @@ gradlePlugin {
       id = 'elasticsearch.validate-rest-spec'
       implementationClass = 'org.elasticsearch.gradle.internal.precommit.ValidateRestSpecPlugin'
     }
+    legacyYamlRestCompatTest {
+      id = 'elasticsearch.legacy-yaml-rest-compat-test'
+      implementationClass = 'org.elasticsearch.gradle.internal.test.rest.compat.compat.LegacyYamlRestCompatTestPlugin'
+    }
     yamlRestCompatTest {
       id = 'elasticsearch.yaml-rest-compat-test'
-      implementationClass = 'org.elasticsearch.gradle.internal.rest.compat.YamlRestCompatTestPlugin'
+      implementationClass = 'org.elasticsearch.gradle.internal.test.rest.compat.compat.YamlRestCompatTestPlugin'
+    }
+    legacyYamlRestTest {
+      id = 'elasticsearch.legacy-yaml-rest-test'
+      implementationClass = 'org.elasticsearch.gradle.internal.test.rest.LegacyYamlRestTestPlugin'
     }
     yamlRestTest {
       id = 'elasticsearch.internal-yaml-rest-test'

build-tools-internal/src/integTest/groovy/org/elasticsearch/gradle/fixtures/AbstractRestResourcesFuncTest.groovy

@@ -13,6 +13,7 @@ abstract class AbstractRestResourcesFuncTest extends AbstractGradleFuncTest {
 
     def setup() {
         subProject(":test:framework") << "apply plugin: 'elasticsearch.java'"
+        subProject(":test:test-clusters") << "apply plugin: 'elasticsearch.java'"
         subProject(":test:yaml-rest-runner") << "apply plugin: 'elasticsearch.java'"
 
         subProject(":rest-api-spec") << """

build-tools-internal/src/integTest/groovy/org/elasticsearch/gradle/internal/precommit/TestingConventionsPrecommitPluginFuncTest.groovy

@@ -174,12 +174,12 @@ class TestingConventionsPrecommitPluginFuncTest extends AbstractGradleInternalPl
         given:
         clazz(dir('src/yamlRestTest/java'), "org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase")
         buildFile << """
-        apply plugin:'elasticsearch.internal-yaml-rest-test'
-        
+        apply plugin:'elasticsearch.legacy-yaml-rest-test'
+
         dependencies {
             yamlRestTestImplementation "org.apache.lucene:tests.util:1.0"
             yamlRestTestImplementation "org.junit:junit:4.42"
-        }    
+        }
         """
 
         clazz(dir("src/yamlRestTest/java"), "org.acme.valid.SomeMatchingIT", "org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase") {
@@ -216,11 +216,11 @@ class TestingConventionsPrecommitPluginFuncTest extends AbstractGradleInternalPl
         buildFile << """
         import org.elasticsearch.gradle.internal.precommit.TestingConventionsCheckTask
         apply plugin:'$pluginName'
-        
+
         dependencies {
             ${sourceSetName}Implementation "org.apache.lucene:tests.util:1.0"
             ${sourceSetName}Implementation "org.junit:junit:4.42"
-        }    
+        }
         tasks.withType(TestingConventionsCheckTask).configureEach {
             suffix 'IT'
             suffix 'Tests'
@@ -252,19 +252,19 @@ class TestingConventionsPrecommitPluginFuncTest extends AbstractGradleInternalPl
         )
 
         where:
-        pluginName                              | taskName                                | sourceSetName
-        "elasticsearch.internal-java-rest-test" | ":javaRestTestTestingConventions"       | "javaRestTest"
+        pluginName                              | taskName                                 | sourceSetName
+        "elasticsearch.legacy-java-rest-test"   | ":javaRestTestTestingConventions"        | "javaRestTest"
         "elasticsearch.internal-cluster-test"   | ":internalClusterTestTestingConventions" | "internalClusterTest"
     }
 
     private void simpleJavaBuild() {
         buildFile << """
         apply plugin:'java'
-                
+
         dependencies {
             testImplementation "org.apache.lucene:tests.util:1.0"
             testImplementation "org.junit:junit:4.42"
-        }    
+        }
         """
     }
 }

build-tools-internal/src/integTest/groovy/org/elasticsearch/gradle/internal/test/rest/YamlRestCompatTestPluginFuncTest.groovy → build-tools-internal/src/integTest/groovy/org/elasticsearch/gradle/internal/test/rest/LegacyYamlRestCompatTestPluginFuncTest.groovy

@@ -17,7 +17,7 @@ import org.elasticsearch.gradle.fixtures.AbstractRestResourcesFuncTest
 import org.elasticsearch.gradle.VersionProperties
 import org.gradle.testkit.runner.TaskOutcome
 
-class YamlRestCompatTestPluginFuncTest extends AbstractRestResourcesFuncTest {
+class LegacyYamlRestCompatTestPluginFuncTest extends AbstractRestResourcesFuncTest {
 
     def compatibleVersion = Version.fromString(VersionProperties.getVersions().get("elasticsearch")).getMajor() - 1
     def specIntermediateDir = "restResources/v${compatibleVersion}/yamlSpecs"
@@ -45,7 +45,7 @@ class YamlRestCompatTestPluginFuncTest extends AbstractRestResourcesFuncTest {
 
         buildFile << """
         plugins {
-          id 'elasticsearch.yaml-rest-compat-test'
+          id 'elasticsearch.legacy-yaml-rest-compat-test'
         }
         """
 
@@ -71,7 +71,7 @@ class YamlRestCompatTestPluginFuncTest extends AbstractRestResourcesFuncTest {
         """
 
         buildFile << """
-            apply plugin: 'elasticsearch.yaml-rest-compat-test'
+            apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
             // avoids a dependency problem in this test, the distribution in use here is inconsequential to the test
             import org.elasticsearch.gradle.testclusters.TestDistribution;
@@ -151,7 +151,7 @@ class YamlRestCompatTestPluginFuncTest extends AbstractRestResourcesFuncTest {
 
         buildFile << """
         plugins {
-          id 'elasticsearch.yaml-rest-compat-test'
+          id 'elasticsearch.legacy-yaml-rest-compat-test'
         }
 
         """
@@ -194,7 +194,7 @@ class YamlRestCompatTestPluginFuncTest extends AbstractRestResourcesFuncTest {
         """
 
         buildFile << """
-            apply plugin: 'elasticsearch.yaml-rest-compat-test'
+            apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
             // avoids a dependency problem in this test, the distribution in use here is inconsequential to the test
             import org.elasticsearch.gradle.testclusters.TestDistribution;

build-tools-internal/src/integTest/groovy/org/elasticsearch/gradle/internal/test/rest/InternalYamlRestTestPluginFuncTest.groovy → build-tools-internal/src/integTest/groovy/org/elasticsearch/gradle/internal/test/rest/LegacyYamlRestTestPluginFuncTest.groovy

@@ -15,7 +15,7 @@ import org.elasticsearch.gradle.fixtures.AbstractRestResourcesFuncTest
 import org.gradle.testkit.runner.TaskOutcome
 
 @IgnoreIf({ os.isWindows() })
-class InternalYamlRestTestPluginFuncTest extends AbstractRestResourcesFuncTest {
+class LegacyYamlRestTestPluginFuncTest extends AbstractRestResourcesFuncTest {
 
     def "yamlRestTest does nothing when there are no tests"() {
         given:
@@ -23,7 +23,7 @@ class InternalYamlRestTestPluginFuncTest extends AbstractRestResourcesFuncTest {
         configurationCacheCompatible = false
         buildFile << """
         plugins {
-          id 'elasticsearch.internal-yaml-rest-test'
+          id 'elasticsearch.legacy-yaml-rest-test'
         }
         """
 
@@ -42,7 +42,7 @@ class InternalYamlRestTestPluginFuncTest extends AbstractRestResourcesFuncTest {
         configurationCacheCompatible = false
         internalBuild()
         buildFile << """
-            apply plugin: 'elasticsearch.internal-yaml-rest-test'
+            apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 
             dependencies {
                yamlRestTestImplementation "junit:junit:4.12"
@@ -96,7 +96,7 @@ class InternalYamlRestTestPluginFuncTest extends AbstractRestResourcesFuncTest {
         def subProjectBuildFile = subProject(pluginProjectPath)
         subProjectBuildFile << """
             apply plugin: 'elasticsearch.esplugin'
-            apply plugin: 'elasticsearch.internal-yaml-rest-test'
+            apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 
             dependencies {
                yamlRestTestImplementation "junit:junit:4.12"

build-tools-internal/src/main/groovy/org/elasticsearch/gradle/internal/doc/DocsTestPlugin.groovy

@@ -38,7 +38,7 @@ class DocsTestPlugin implements Plugin<Project> {
 
     @Override
     void apply(Project project) {
-        project.pluginManager.apply('elasticsearch.internal-yaml-rest-test')
+        project.pluginManager.apply('elasticsearch.legacy-yaml-rest-test')
 
         String distribution = System.getProperty('tests.distribution', 'default')
         // The distribution can be configured with -Dtests.distribution on the command line

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/ResolveAllDependencies.java

@@ -21,7 +21,7 @@ import java.util.stream.Collectors;
 import javax.inject.Inject;
 
 import static org.elasticsearch.gradle.DistributionDownloadPlugin.DISTRO_EXTRACTED_CONFIG_PREFIX;
-import static org.elasticsearch.gradle.internal.rest.compat.YamlRestCompatTestPlugin.BWC_MINOR_CONFIG_NAME;
+import static org.elasticsearch.gradle.internal.test.rest.compat.compat.LegacyYamlRestCompatTestPlugin.BWC_MINOR_CONFIG_NAME;
 
 public class ResolveAllDependencies extends DefaultTask {
 

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/precommit/TestingConventionsPrecommitPlugin.java

@@ -11,7 +11,8 @@ package org.elasticsearch.gradle.internal.precommit;
 import org.elasticsearch.gradle.internal.conventions.precommit.PrecommitPlugin;
 import org.elasticsearch.gradle.internal.test.InternalClusterTestPlugin;
 import org.elasticsearch.gradle.internal.test.rest.InternalJavaRestTestPlugin;
-import org.elasticsearch.gradle.internal.test.rest.InternalYamlRestTestPlugin;
+import org.elasticsearch.gradle.internal.test.rest.LegacyJavaRestTestPlugin;
+import org.elasticsearch.gradle.internal.test.rest.LegacyYamlRestTestPlugin;
 import org.gradle.api.Action;
 import org.gradle.api.NamedDomainObjectProvider;
 import org.gradle.api.Project;
@@ -43,8 +44,8 @@ public class TestingConventionsPrecommitPlugin extends PrecommitPlugin {
             });
         });
 
-        project.getPlugins().withType(InternalYamlRestTestPlugin.class, yamlRestTestPlugin -> {
-            NamedDomainObjectProvider<SourceSet> sourceSet = sourceSets.named(InternalYamlRestTestPlugin.SOURCE_SET_NAME);
+        project.getPlugins().withType(LegacyYamlRestTestPlugin.class, yamlRestTestPlugin -> {
+            NamedDomainObjectProvider<SourceSet> sourceSet = sourceSets.named(LegacyYamlRestTestPlugin.SOURCE_SET_NAME);
             setupTaskForSourceSet(project, sourceSet, t -> {
                 t.getSuffixes().convention(List.of("IT"));
                 t.getBaseClasses().convention(List.of("org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase"));
@@ -68,8 +69,17 @@ public class TestingConventionsPrecommitPlugin extends PrecommitPlugin {
             });
         });
 
+        project.getPlugins().withType(LegacyJavaRestTestPlugin.class, javaRestTestPlugin -> {
+            NamedDomainObjectProvider<SourceSet> sourceSet = sourceSets.named(LegacyJavaRestTestPlugin.SOURCE_SET_NAME);
+            setupTaskForSourceSet(project, sourceSet, t -> {
+                t.getSuffixes().convention(List.of("IT"));
+                t.getBaseClasses()
+                    .convention(List.of("org.elasticsearch.test.ESIntegTestCase", "org.elasticsearch.test.rest.ESRestTestCase"));
+            });
+        });
+
         project.getPlugins().withType(InternalJavaRestTestPlugin.class, javaRestTestPlugin -> {
-            NamedDomainObjectProvider<SourceSet> sourceSet = sourceSets.named(InternalJavaRestTestPlugin.SOURCE_SET_NAME);
+            NamedDomainObjectProvider<SourceSet> sourceSet = sourceSets.named(LegacyJavaRestTestPlugin.SOURCE_SET_NAME);
             setupTaskForSourceSet(project, sourceSet, t -> {
                 t.getSuffixes().convention(List.of("IT"));
                 t.getBaseClasses()

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/RestTestBasePlugin.java → build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/LegacyRestTestBasePlugin.java

@@ -31,7 +31,11 @@ import javax.inject.Inject;
 import static org.elasticsearch.gradle.plugin.BasePluginBuildPlugin.BUNDLE_PLUGIN_TASK_NAME;
 import static org.elasticsearch.gradle.plugin.BasePluginBuildPlugin.EXPLODED_BUNDLE_PLUGIN_TASK_NAME;
 
-public class RestTestBasePlugin implements Plugin<Project> {
+/**
+ * @deprecated use {@link RestTestBasePlugin} instead
+ */
+@Deprecated
+public class LegacyRestTestBasePlugin implements Plugin<Project> {
     private static final String TESTS_REST_CLUSTER = "tests.rest.cluster";
     private static final String TESTS_CLUSTER = "tests.cluster";
     private static final String TESTS_CLUSTER_NAME = "tests.clustername";
@@ -40,7 +44,7 @@ public class RestTestBasePlugin implements Plugin<Project> {
     private ProviderFactory providerFactory;
 
     @Inject
-    public RestTestBasePlugin(ProviderFactory providerFactory) {
+    public LegacyRestTestBasePlugin(ProviderFactory providerFactory) {
         this.providerFactory = providerFactory;
     }
 
@@ -87,17 +91,19 @@ public class RestTestBasePlugin implements Plugin<Project> {
             .withType(StandaloneRestIntegTestTask.class)
             .configureEach(t -> t.finalizedBy(project.getTasks().withType(FixtureStop.class)));
 
-        project.getTasks().withType(StandaloneRestIntegTestTask.class).configureEach(t ->
-        // if this a module or plugin, it may have an associated zip file with it's contents, add that to the test cluster
-        project.getPluginManager().withPlugin("elasticsearch.esplugin", plugin -> {
-            if (GradleUtils.isModuleProject(project.getPath())) {
-                var bundle = project.getTasks().withType(Sync.class).named(EXPLODED_BUNDLE_PLUGIN_TASK_NAME);
-                t.getClusters().forEach(c -> c.module(bundle));
-            } else {
-                var bundle = project.getTasks().withType(Zip.class).named(BUNDLE_PLUGIN_TASK_NAME);
-                t.getClusters().forEach(c -> c.plugin(bundle));
-            }
-        }));
+        project.getTasks().withType(StandaloneRestIntegTestTask.class).configureEach(t -> {
+            t.setMaxParallelForks(1);
+            // if this a module or plugin, it may have an associated zip file with it's contents, add that to the test cluster
+            project.getPluginManager().withPlugin("elasticsearch.esplugin", plugin -> {
+                if (GradleUtils.isModuleProject(project.getPath())) {
+                    var bundle = project.getTasks().withType(Sync.class).named(EXPLODED_BUNDLE_PLUGIN_TASK_NAME);
+                    t.getClusters().forEach(c -> c.module(bundle));
+                } else {
+                    var bundle = project.getTasks().withType(Zip.class).named(BUNDLE_PLUGIN_TASK_NAME);
+                    t.getClusters().forEach(c -> c.plugin(bundle));
+                }
+            });
+        });
     }
 
     private String systemProperty(String propName) {

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/StandaloneRestTestPlugin.java

@@ -11,8 +11,8 @@ package org.elasticsearch.gradle.internal.test;
 import org.elasticsearch.gradle.internal.ExportElasticsearchBuildResourcesTask;
 import org.elasticsearch.gradle.internal.info.GlobalBuildInfoPlugin;
 import org.elasticsearch.gradle.internal.precommit.InternalPrecommitTasks;
-import org.elasticsearch.gradle.internal.test.rest.InternalJavaRestTestPlugin;
-import org.elasticsearch.gradle.internal.test.rest.InternalYamlRestTestPlugin;
+import org.elasticsearch.gradle.internal.test.rest.LegacyJavaRestTestPlugin;
+import org.elasticsearch.gradle.internal.test.rest.LegacyYamlRestTestPlugin;
 import org.elasticsearch.gradle.internal.test.rest.RestTestUtil;
 import org.gradle.api.InvalidUserDataException;
 import org.gradle.api.Plugin;
@@ -32,8 +32,8 @@ import java.util.Map;
  * and run REST tests. Use BuildPlugin if you want to build main code as well
  * as tests.
  *
- * @deprecated use {@link InternalClusterTestPlugin}, {@link InternalJavaRestTestPlugin} or
- * {@link InternalYamlRestTestPlugin} instead.
+ * @deprecated use {@link InternalClusterTestPlugin}, {@link LegacyJavaRestTestPlugin} or
+ * {@link LegacyYamlRestTestPlugin} instead.
  */
 @Deprecated
 public class StandaloneRestTestPlugin implements Plugin<Project> {
@@ -46,7 +46,7 @@ public class StandaloneRestTestPlugin implements Plugin<Project> {
         }
 
         project.getRootProject().getPluginManager().apply(GlobalBuildInfoPlugin.class);
-        project.getPluginManager().apply(RestTestBasePlugin.class);
+        project.getPluginManager().apply(LegacyRestTestBasePlugin.class);
 
         project.getTasks().register("buildResources", ExportElasticsearchBuildResourcesTask.class);
 

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/TestWithSslPlugin.java

@@ -14,7 +14,7 @@ import org.elasticsearch.gradle.internal.info.BuildParams;
 import org.elasticsearch.gradle.internal.precommit.FilePermissionsPrecommitPlugin;
 import org.elasticsearch.gradle.internal.precommit.ForbiddenPatternsPrecommitPlugin;
 import org.elasticsearch.gradle.internal.precommit.ForbiddenPatternsTask;
-import org.elasticsearch.gradle.internal.test.rest.InternalJavaRestTestPlugin;
+import org.elasticsearch.gradle.internal.test.rest.LegacyJavaRestTestPlugin;
 import org.elasticsearch.gradle.testclusters.ElasticsearchCluster;
 import org.elasticsearch.gradle.testclusters.TestClustersAware;
 import org.elasticsearch.gradle.testclusters.TestClustersPlugin;
@@ -62,8 +62,8 @@ public class TestWithSslPlugin implements Plugin<Project> {
                 .withType(RestIntegTestTask.class)
                 .configureEach(runner -> runner.systemProperty("tests.ssl.enabled", "true"));
         });
-        project.getPlugins().withType(InternalJavaRestTestPlugin.class).configureEach(restTestPlugin -> {
-            SourceSet testSourceSet = Util.getJavaSourceSets(project).getByName(InternalJavaRestTestPlugin.SOURCE_SET_NAME);
+        project.getPlugins().withType(LegacyJavaRestTestPlugin.class).configureEach(restTestPlugin -> {
+            SourceSet testSourceSet = Util.getJavaSourceSets(project).getByName(LegacyJavaRestTestPlugin.SOURCE_SET_NAME);
             testSourceSet.getResources().srcDir(new File(keyStoreDir, "test/ssl"));
             project.getTasks().named(testSourceSet.getProcessResourcesTaskName()).configure(t -> t.dependsOn(exportKeyStore));
             project.getTasks().withType(TestClustersAware.class).configureEach(clusterAware -> clusterAware.dependsOn(exportKeyStore));

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/rest/InternalJavaRestTestPlugin.java

@@ -8,7 +8,7 @@
 
 package org.elasticsearch.gradle.internal.test.rest;
 
-import org.elasticsearch.gradle.internal.test.RestTestBasePlugin;
+import org.elasticsearch.gradle.testclusters.StandaloneRestIntegTestTask;
 import org.elasticsearch.gradle.util.GradleUtils;
 import org.gradle.api.Plugin;
 import org.gradle.api.Project;
@@ -33,8 +33,11 @@ public class InternalJavaRestTestPlugin implements Plugin<Project> {
         SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
         SourceSet javaTestSourceSet = sourceSets.create(SOURCE_SET_NAME);
 
+        project.getDependencies().add(javaTestSourceSet.getImplementationConfigurationName(), project.project(":test:test-clusters"));
+
         // setup the javaRestTest task
-        registerTestTask(project, javaTestSourceSet);
+        // we use a StandloneRestIntegTestTask here so that the conventions of RestTestBasePlugin don't create a test cluster
+        registerTestTask(project, javaTestSourceSet, SOURCE_SET_NAME, StandaloneRestIntegTestTask.class);
 
         // setup dependencies
         setupJavaRestTestDependenciesDefaults(project, javaTestSourceSet);

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/rest/InternalYamlRestTestPlugin.java

@@ -8,7 +8,7 @@
 
 package org.elasticsearch.gradle.internal.test.rest;
 
-import org.elasticsearch.gradle.internal.test.RestTestBasePlugin;
+import org.elasticsearch.gradle.testclusters.StandaloneRestIntegTestTask;
 import org.elasticsearch.gradle.util.GradleUtils;
 import org.gradle.api.Plugin;
 import org.gradle.api.Project;
@@ -34,10 +34,10 @@ public class InternalYamlRestTestPlugin implements Plugin<Project> {
         SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
         SourceSet yamlTestSourceSet = sourceSets.create(SOURCE_SET_NAME);
 
-        registerTestTask(project, yamlTestSourceSet);
+        registerTestTask(project, yamlTestSourceSet, SOURCE_SET_NAME, StandaloneRestIntegTestTask.class);
 
         // setup the dependencies
-        setupYamlRestTestDependenciesDefaults(project, yamlTestSourceSet);
+        setupYamlRestTestDependenciesDefaults(project, yamlTestSourceSet, true);
 
         // setup the copy for the rest resources
         project.getTasks().withType(CopyRestApiTask.class).configureEach(copyRestApiTask -> {

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/rest/LegacyJavaRestTestPlugin.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.gradle.internal.test.rest;
+
+import org.elasticsearch.gradle.internal.test.LegacyRestTestBasePlugin;
+import org.elasticsearch.gradle.util.GradleUtils;
+import org.gradle.api.Plugin;
+import org.gradle.api.Project;
+import org.gradle.api.tasks.SourceSet;
+import org.gradle.api.tasks.SourceSetContainer;
+
+import static org.elasticsearch.gradle.internal.test.rest.RestTestUtil.registerTestTask;
+import static org.elasticsearch.gradle.internal.test.rest.RestTestUtil.setupJavaRestTestDependenciesDefaults;
+
+/**
+ * Apply this plugin to run the Java based REST tests.
+ *
+ * @deprecated use {@link InternalJavaRestTestPlugin}
+ */
+@Deprecated
+public class LegacyJavaRestTestPlugin implements Plugin<Project> {
+
+    public static final String SOURCE_SET_NAME = "javaRestTest";
+
+    @Override
+    public void apply(Project project) {
+        project.getPluginManager().apply(LegacyRestTestBasePlugin.class);
+
+        // create source set
+        SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
+        SourceSet javaTestSourceSet = sourceSets.create(SOURCE_SET_NAME);
+
+        // setup the javaRestTest task
+        registerTestTask(project, javaTestSourceSet);
+
+        // setup dependencies
+        setupJavaRestTestDependenciesDefaults(project, javaTestSourceSet);
+
+        // setup IDE
+        GradleUtils.setupIdeForTestSourceSet(project, javaTestSourceSet);
+    }
+}

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/rest/LegacyYamlRestTestPlugin.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.gradle.internal.test.rest;
+
+import org.elasticsearch.gradle.internal.test.LegacyRestTestBasePlugin;
+import org.elasticsearch.gradle.util.GradleUtils;
+import org.gradle.api.Plugin;
+import org.gradle.api.Project;
+import org.gradle.api.tasks.SourceSet;
+import org.gradle.api.tasks.SourceSetContainer;
+
+import static org.elasticsearch.gradle.internal.test.rest.RestTestUtil.registerTestTask;
+import static org.elasticsearch.gradle.internal.test.rest.RestTestUtil.setupYamlRestTestDependenciesDefaults;
+
+/**
+ * Apply this plugin to run the YAML based REST tests.
+ *
+ * @deprecated use {@link InternalYamlRestTestPlugin}
+ */
+@Deprecated
+public class LegacyYamlRestTestPlugin implements Plugin<Project> {
+
+    public static final String SOURCE_SET_NAME = "yamlRestTest";
+
+    @Override
+    public void apply(Project project) {
+        project.getPluginManager().apply(LegacyRestTestBasePlugin.class);
+        project.getPluginManager().apply(RestResourcesPlugin.class);
+
+        // create source set
+        SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
+        SourceSet yamlTestSourceSet = sourceSets.create(SOURCE_SET_NAME);
+
+        registerTestTask(project, yamlTestSourceSet);
+
+        // setup the dependencies
+        setupYamlRestTestDependenciesDefaults(project, yamlTestSourceSet);
+
+        // setup the copy for the rest resources
+        project.getTasks().withType(CopyRestApiTask.class).configureEach(copyRestApiTask -> {
+            copyRestApiTask.setSourceResourceDir(
+                yamlTestSourceSet.getResources()
+                    .getSrcDirs()
+                    .stream()
+                    .filter(f -> f.isDirectory() && f.getName().equals("resources"))
+                    .findFirst()
+                    .orElse(null)
+            );
+        });
+
+        // Register rest resources with source set
+        yamlTestSourceSet.getOutput()
+            .dir(
+                project.getTasks()
+                    .withType(CopyRestApiTask.class)
+                    .named(RestResourcesPlugin.COPY_REST_API_SPECS_TASK)
+                    .flatMap(CopyRestApiTask::getOutputResourceDir)
+            );
+
+        yamlTestSourceSet.getOutput()
+            .dir(
+                project.getTasks()
+                    .withType(CopyRestTestsTask.class)
+                    .named(RestResourcesPlugin.COPY_YAML_TESTS_TASK)
+                    .flatMap(CopyRestTestsTask::getOutputResourceDir)
+            );
+
+        GradleUtils.setupIdeForTestSourceSet(project, yamlTestSourceSet);
+    }
+}

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/rest/RestTestBasePlugin.java

@@ -0,0 +1,227 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.gradle.internal.test.rest;
+
+import groovy.lang.Closure;
+
+import org.elasticsearch.gradle.Architecture;
+import org.elasticsearch.gradle.DistributionDownloadPlugin;
+import org.elasticsearch.gradle.ElasticsearchDistribution;
+import org.elasticsearch.gradle.VersionProperties;
+import org.elasticsearch.gradle.distribution.ElasticsearchDistributionTypes;
+import org.elasticsearch.gradle.internal.ElasticsearchJavaPlugin;
+import org.elasticsearch.gradle.internal.InternalDistributionDownloadPlugin;
+import org.elasticsearch.gradle.internal.info.BuildParams;
+import org.elasticsearch.gradle.plugin.PluginBuildPlugin;
+import org.elasticsearch.gradle.plugin.PluginPropertiesExtension;
+import org.elasticsearch.gradle.test.SystemPropertyCommandLineArgumentProvider;
+import org.elasticsearch.gradle.testclusters.StandaloneRestIntegTestTask;
+import org.elasticsearch.gradle.transform.UnzipTransform;
+import org.elasticsearch.gradle.util.GradleUtils;
+import org.gradle.api.Action;
+import org.gradle.api.NamedDomainObjectContainer;
+import org.gradle.api.Plugin;
+import org.gradle.api.Project;
+import org.gradle.api.Task;
+import org.gradle.api.artifacts.Configuration;
+import org.gradle.api.artifacts.Dependency;
+import org.gradle.api.artifacts.ProjectDependency;
+import org.gradle.api.artifacts.type.ArtifactTypeDefinition;
+import org.gradle.api.file.FileTree;
+import org.gradle.api.plugins.JavaBasePlugin;
+import org.gradle.api.provider.ProviderFactory;
+import org.gradle.api.tasks.ClasspathNormalizer;
+import org.gradle.api.tasks.PathSensitivity;
+import org.gradle.api.tasks.util.PatternFilterable;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import javax.inject.Inject;
+
+/**
+ * Base plugin used for wiring up build tasks to REST testing tasks using new JUnit rule-based test clusters framework.
+ */
+public class RestTestBasePlugin implements Plugin<Project> {
+
+    private static final String TESTS_RUNTIME_JAVA_SYSPROP = "tests.runtime.java";
+    private static final String DEFAULT_DISTRIBUTION_SYSPROP = "tests.default.distribution";
+    private static final String INTEG_TEST_DISTRIBUTION_SYSPROP = "tests.integ-test.distribution";
+    private static final String TESTS_CLUSTER_MODULES_PATH_SYSPROP = "tests.cluster.modules.path";
+    private static final String TESTS_CLUSTER_PLUGINS_PATH_SYSPROP = "tests.cluster.plugins.path";
+    private static final String DEFAULT_REST_INTEG_TEST_DISTRO = "default_distro";
+    private static final String INTEG_TEST_REST_INTEG_TEST_DISTRO = "integ_test_distro";
+    private static final String MODULES_CONFIGURATION = "clusterModules";
+    private static final String PLUGINS_CONFIGURATION = "clusterPlugins";
+    private static final String EXTRACTED_PLUGINS_CONFIGURATION = "extractedPlugins";
+
+    private final ProviderFactory providerFactory;
+
+    @Inject
+    public RestTestBasePlugin(ProviderFactory providerFactory) {
+        this.providerFactory = providerFactory;
+    }
+
+    @Override
+    public void apply(Project project) {
+        project.getPluginManager().apply(ElasticsearchJavaPlugin.class);
+        project.getPluginManager().apply(InternalDistributionDownloadPlugin.class);
+
+        // Register integ-test and default distributions
+        NamedDomainObjectContainer<ElasticsearchDistribution> distributions = DistributionDownloadPlugin.getContainer(project);
+        ElasticsearchDistribution defaultDistro = distributions.create(DEFAULT_REST_INTEG_TEST_DISTRO, distro -> {
+            distro.setVersion(VersionProperties.getElasticsearch());
+            distro.setArchitecture(Architecture.current());
+        });
+        ElasticsearchDistribution integTestDistro = distributions.create(INTEG_TEST_REST_INTEG_TEST_DISTRO, distro -> {
+            distro.setVersion(VersionProperties.getElasticsearch());
+            distro.setArchitecture(Architecture.current());
+            distro.setType(ElasticsearchDistributionTypes.INTEG_TEST_ZIP);
+        });
+
+        // Create configures for module and plugin dependencies
+        Configuration modulesConfiguration = createPluginConfiguration(project, MODULES_CONFIGURATION, true);
+        Configuration pluginsConfiguration = createPluginConfiguration(project, PLUGINS_CONFIGURATION, false);
+        Configuration extractedPluginsConfiguration = createPluginConfiguration(project, EXTRACTED_PLUGINS_CONFIGURATION, true);
+        extractedPluginsConfiguration.extendsFrom(pluginsConfiguration);
+        configureArtifactTransforms(project);
+
+        // For plugin and module projects, register the current project plugin bundle as a dependency
+        project.getPluginManager().withPlugin("elasticsearch.esplugin", plugin -> {
+            if (GradleUtils.isModuleProject(project.getPath())) {
+                project.getDependencies()
+                    .add(modulesConfiguration.getName(), project.getDependencies().project(Map.of("path", project.getPath())));
+            } else {
+                project.getDependencies().add(pluginsConfiguration.getName(), project.files(project.getTasks().named("bundlePlugin")));
+            }
+
+        });
+
+        project.getTasks().withType(StandaloneRestIntegTestTask.class, task -> {
+            SystemPropertyCommandLineArgumentProvider nonInputSystemProperties = task.getExtensions()
+                .getByType(SystemPropertyCommandLineArgumentProvider.class);
+
+            task.dependsOn(integTestDistro, modulesConfiguration);
+            registerDistributionInputs(task, integTestDistro);
+
+            task.setMaxParallelForks(task.getProject().getGradle().getStartParameter().getMaxWorkerCount() / 2);
+
+            // Disable the security manager and syscall filter since the test framework needs to fork processes
+            task.systemProperty("tests.security.manager", "false");
+            task.systemProperty("tests.system_call_filter", "false");
+
+            // Register plugins and modules as task inputs and pass paths as system properties to tests
+            nonInputSystemProperties.systemProperty(TESTS_CLUSTER_MODULES_PATH_SYSPROP, modulesConfiguration::getAsPath);
+            registerConfigurationInputs(task, modulesConfiguration);
+            nonInputSystemProperties.systemProperty(TESTS_CLUSTER_PLUGINS_PATH_SYSPROP, pluginsConfiguration::getAsPath);
+            registerConfigurationInputs(task, extractedPluginsConfiguration);
+
+            // Wire up integ-test distribution by default for all test tasks
+            nonInputSystemProperties.systemProperty(
+                INTEG_TEST_DISTRIBUTION_SYSPROP,
+                () -> integTestDistro.getExtracted().getSingleFile().getPath()
+            );
+            nonInputSystemProperties.systemProperty(TESTS_RUNTIME_JAVA_SYSPROP, BuildParams.getRuntimeJavaHome());
+
+            // Add `usesDefaultDistribution()` extension method to test tasks to indicate they require the default distro
+            task.getExtensions().getExtraProperties().set("usesDefaultDistribution", new Closure<Void>(task) {
+                @Override
+                public Void call(Object... args) {
+                    task.dependsOn(defaultDistro);
+                    registerDistributionInputs(task, defaultDistro);
+
+                    nonInputSystemProperties.systemProperty(
+                        DEFAULT_DISTRIBUTION_SYSPROP,
+                        providerFactory.provider(() -> defaultDistro.getExtracted().getSingleFile().getPath())
+                    );
+                    return null;
+                }
+            });
+        });
+
+        project.getTasks()
+            .named(JavaBasePlugin.CHECK_TASK_NAME)
+            .configure(check -> check.dependsOn(project.getTasks().withType(StandaloneRestIntegTestTask.class)));
+    }
+
+    private FileTree getDistributionFiles(ElasticsearchDistribution distribution, Action<PatternFilterable> patternFilter) {
+        return distribution.getExtracted().getAsFileTree().matching(patternFilter);
+    }
+
+    private void registerConfigurationInputs(Task task, Configuration configuration) {
+        task.getInputs()
+            .files(providerFactory.provider(() -> configuration.getAsFileTree().filter(f -> f.getName().endsWith(".jar"))))
+            .withPropertyName(configuration.getName() + "-classpath")
+            .withNormalizer(ClasspathNormalizer.class);
+
+        task.getInputs()
+            .files(providerFactory.provider(() -> configuration.getAsFileTree().filter(f -> f.getName().endsWith(".jar") == false)))
+            .withPropertyName(configuration.getName() + "-files")
+            .withPathSensitivity(PathSensitivity.RELATIVE);
+    }
+
+    private void registerDistributionInputs(Task task, ElasticsearchDistribution distribution) {
+        task.getInputs()
+            .files(providerFactory.provider(() -> getDistributionFiles(distribution, filter -> filter.exclude("**/*.jar"))))
+            .withPropertyName(distribution.getName() + "-files")
+            .withPathSensitivity(PathSensitivity.RELATIVE);
+
+        task.getInputs()
+            .files(providerFactory.provider(() -> getDistributionFiles(distribution, filter -> filter.include("**/*.jar"))))
+            .withPropertyName(distribution.getName() + "-classpath")
+            .withNormalizer(ClasspathNormalizer.class);
+    }
+
+    private Optional<String> findModulePath(Project project, String pluginName) {
+        return project.getRootProject()
+            .getAllprojects()
+            .stream()
+            .filter(p -> GradleUtils.isModuleProject(p.getPath()))
+            .filter(p -> p.getPlugins().hasPlugin(PluginBuildPlugin.class))
+            .filter(p -> p.getExtensions().getByType(PluginPropertiesExtension.class).getName().equals(pluginName))
+            .findFirst()
+            .map(Project::getPath);
+    }
+
+    private Configuration createPluginConfiguration(Project project, String name, boolean useExploded) {
+        return project.getConfigurations().create(name, c -> {
+            if (useExploded) {
+                c.attributes(a -> a.attribute(ArtifactTypeDefinition.ARTIFACT_TYPE_ATTRIBUTE, ArtifactTypeDefinition.DIRECTORY_TYPE));
+            } else {
+                c.attributes(a -> a.attribute(ArtifactTypeDefinition.ARTIFACT_TYPE_ATTRIBUTE, ArtifactTypeDefinition.ZIP_TYPE));
+            }
+            c.withDependencies(dependencies -> {
+                // Add dependencies of any modules
+                for (Dependency dependency : dependencies) {
+                    if (dependency instanceof ProjectDependency projectDependency) {
+                        List<String> extendedPlugins = projectDependency.getDependencyProject()
+                            .getExtensions()
+                            .getByType(PluginPropertiesExtension.class)
+                            .getExtendedPlugins();
+
+                        for (String extendedPlugin : extendedPlugins) {
+                            findModulePath(project, extendedPlugin).ifPresent(
+                                modulePath -> dependencies.add(project.getDependencies().project(Map.of("path", modulePath)))
+                            );
+                        }
+                    }
+                }
+            });
+        });
+    }
+
+    private void configureArtifactTransforms(Project project) {
+        project.getDependencies().registerTransform(UnzipTransform.class, transformSpec -> {
+            transformSpec.getFrom().attribute(ArtifactTypeDefinition.ARTIFACT_TYPE_ATTRIBUTE, ArtifactTypeDefinition.ZIP_TYPE);
+            transformSpec.getTo().attribute(ArtifactTypeDefinition.ARTIFACT_TYPE_ATTRIBUTE, ArtifactTypeDefinition.DIRECTORY_TYPE);
+            transformSpec.getParameters().setAsFiletreeOutput(false);
+        });
+    }
+}

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/rest/RestTestUtil.java

@@ -15,6 +15,7 @@ import org.gradle.api.plugins.JavaPlugin;
 import org.gradle.api.provider.Provider;
 import org.gradle.api.tasks.SourceSet;
 import org.gradle.api.tasks.TaskProvider;
+import org.gradle.api.tasks.testing.Test;
 
 /**
  * Utility class to configure the necessary tasks and dependencies.
@@ -34,8 +35,17 @@ public class RestTestUtil {
      * Creates a {@link RestIntegTestTask} task with a custom name for the provided source set
      */
     public static TaskProvider<RestIntegTestTask> registerTestTask(Project project, SourceSet sourceSet, String taskName) {
+        return registerTestTask(project, sourceSet, taskName, RestIntegTestTask.class);
+    }
+
+    /**
+     * Creates a {@link T} task with a custom name for the provided source set
+     *
+     * @param <T> test task type
+     */
+    public static <T extends Test> TaskProvider<T> registerTestTask(Project project, SourceSet sourceSet, String taskName, Class<T> clazz) {
         // lazily create the test task
-        return project.getTasks().register(taskName, RestIntegTestTask.class, testTask -> {
+        return project.getTasks().register(taskName, clazz, testTask -> {
             testTask.setGroup(JavaBasePlugin.VERIFICATION_GROUP);
             testTask.setDescription("Runs the REST tests against an external cluster");
             project.getPlugins().withType(JavaPlugin.class, t -> testTask.mustRunAfter(project.getTasks().named("test")));
@@ -49,10 +59,20 @@ public class RestTestUtil {
      * Setup the dependencies needed for the YAML REST tests.
      */
     public static void setupYamlRestTestDependenciesDefaults(Project project, SourceSet sourceSet) {
+        setupYamlRestTestDependenciesDefaults(project, sourceSet, false);
+    }
+
+    /**
+     * Setup the dependencies needed for the YAML REST tests.
+     */
+    public static void setupYamlRestTestDependenciesDefaults(Project project, SourceSet sourceSet, boolean useNewTestClusters) {
         Project yamlTestRunnerProject = project.findProject(":test:yaml-rest-runner");
         // we shield the project dependency to make integration tests easier
         if (yamlTestRunnerProject != null) {
             project.getDependencies().add(sourceSet.getImplementationConfigurationName(), yamlTestRunnerProject);
+            if (useNewTestClusters) {
+                project.getDependencies().add(sourceSet.getImplementationConfigurationName(), project.project(":test:test-clusters"));
+            }
         }
     }
 

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/rest/compat/YamlRestCompatTestPlugin.java → build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/rest/compat/compat/AbstractYamlRestCompatTestPlugin.java

@@ -6,19 +6,17 @@
  * Side Public License, v 1.
  */
 
-package org.elasticsearch.gradle.internal.rest.compat;
+package org.elasticsearch.gradle.internal.test.rest.compat.compat;
 
 import org.elasticsearch.gradle.Version;
 import org.elasticsearch.gradle.VersionProperties;
 import org.elasticsearch.gradle.internal.ElasticsearchJavaBasePlugin;
-import org.elasticsearch.gradle.internal.test.RestIntegTestTask;
-import org.elasticsearch.gradle.internal.test.RestTestBasePlugin;
+import org.elasticsearch.gradle.internal.test.LegacyRestTestBasePlugin;
 import org.elasticsearch.gradle.internal.test.rest.CopyRestApiTask;
 import org.elasticsearch.gradle.internal.test.rest.CopyRestTestsTask;
-import org.elasticsearch.gradle.internal.test.rest.InternalYamlRestTestPlugin;
+import org.elasticsearch.gradle.internal.test.rest.LegacyYamlRestTestPlugin;
 import org.elasticsearch.gradle.internal.test.rest.RestResourcesExtension;
 import org.elasticsearch.gradle.internal.test.rest.RestResourcesPlugin;
-import org.elasticsearch.gradle.internal.test.rest.RestTestUtil;
 import org.elasticsearch.gradle.testclusters.TestClustersPlugin;
 import org.elasticsearch.gradle.util.GradleUtils;
 import org.gradle.api.Plugin;
@@ -37,6 +35,7 @@ import org.gradle.api.tasks.SourceSet;
 import org.gradle.api.tasks.SourceSetContainer;
 import org.gradle.api.tasks.Sync;
 import org.gradle.api.tasks.TaskProvider;
+import org.gradle.api.tasks.testing.Test;
 
 import java.io.File;
 import java.nio.file.Path;
@@ -50,7 +49,7 @@ import static org.elasticsearch.gradle.internal.test.rest.RestTestUtil.setupYaml
 /**
  * Apply this plugin to run the YAML based REST tests from a prior major version against this version's cluster.
  */
-public class YamlRestCompatTestPlugin implements Plugin<Project> {
+public abstract class AbstractYamlRestCompatTestPlugin implements Plugin<Project> {
     public static final String BWC_MINOR_CONFIG_NAME = "bwcMinor";
     private static final String REST_COMPAT_CHECK_TASK_NAME = "checkRestCompat";
     private static final String COMPATIBILITY_APIS_CONFIGURATION = "restCompatSpecs";
@@ -67,7 +66,7 @@ public class YamlRestCompatTestPlugin implements Plugin<Project> {
     private FileOperations fileOperations;
 
     @Inject
-    public YamlRestCompatTestPlugin(ProjectLayout projectLayout, FileOperations fileOperations) {
+    public AbstractYamlRestCompatTestPlugin(ProjectLayout projectLayout, FileOperations fileOperations) {
         this.projectLayout = projectLayout;
         this.fileOperations = fileOperations;
     }
@@ -81,17 +80,17 @@ public class YamlRestCompatTestPlugin implements Plugin<Project> {
 
         project.getPluginManager().apply(ElasticsearchJavaBasePlugin.class);
         project.getPluginManager().apply(TestClustersPlugin.class);
-        project.getPluginManager().apply(RestTestBasePlugin.class);
+        project.getPluginManager().apply(LegacyRestTestBasePlugin.class);
         project.getPluginManager().apply(RestResourcesPlugin.class);
-        project.getPluginManager().apply(InternalYamlRestTestPlugin.class);
+        project.getPluginManager().apply(getBasePlugin());
 
         RestResourcesExtension extension = project.getExtensions().getByType(RestResourcesExtension.class);
 
         // create source set
         SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
         SourceSet yamlCompatTestSourceSet = sourceSets.create(SOURCE_SET_NAME);
-        SourceSet yamlTestSourceSet = sourceSets.getByName(InternalYamlRestTestPlugin.SOURCE_SET_NAME);
-        GradleUtils.extendSourceSet(project, InternalYamlRestTestPlugin.SOURCE_SET_NAME, SOURCE_SET_NAME);
+        SourceSet yamlTestSourceSet = sourceSets.getByName(LegacyYamlRestTestPlugin.SOURCE_SET_NAME);
+        GradleUtils.extendSourceSet(project, LegacyYamlRestTestPlugin.SOURCE_SET_NAME, SOURCE_SET_NAME);
 
         // copy compatible rest specs
         Configuration bwcMinorConfig = project.getConfigurations().create(BWC_MINOR_CONFIG_NAME);
@@ -217,11 +216,9 @@ public class YamlRestCompatTestPlugin implements Plugin<Project> {
             .named(RestResourcesPlugin.COPY_YAML_TESTS_TASK)
             .flatMap(CopyRestTestsTask::getOutputResourceDir);
 
-        String testTaskName = "yamlRestTestV" + COMPATIBLE_VERSION + "CompatTest";
-
         // setup the test task
-        Provider<RestIntegTestTask> yamlRestCompatTestTask = RestTestUtil.registerTestTask(project, yamlCompatTestSourceSet, testTaskName);
-        project.getTasks().withType(RestIntegTestTask.class).named(testTaskName).configure(testTask -> {
+        TaskProvider<? extends Test> yamlRestCompatTestTask = registerTestTask(project, yamlCompatTestSourceSet);
+        yamlRestCompatTestTask.configure(testTask -> {
             testTask.systemProperty("tests.restCompat", true);
             // Use test runner and classpath from "normal" yaml source set
             testTask.setTestClassesDirs(
@@ -236,11 +233,11 @@ public class YamlRestCompatTestPlugin implements Plugin<Project> {
             );
 
             // run compatibility tests after "normal" tests
-            testTask.mustRunAfter(project.getTasks().named(InternalYamlRestTestPlugin.SOURCE_SET_NAME));
+            testTask.mustRunAfter(project.getTasks().named(LegacyYamlRestTestPlugin.SOURCE_SET_NAME));
             testTask.onlyIf(t -> isEnabled(extraProperties));
         });
 
-        setupYamlRestTestDependenciesDefaults(project, yamlCompatTestSourceSet);
+        setupYamlRestTestDependenciesDefaults(project, yamlCompatTestSourceSet, true);
 
         // setup IDE
         GradleUtils.setupIdeForTestSourceSet(project, yamlCompatTestSourceSet);
@@ -259,6 +256,10 @@ public class YamlRestCompatTestPlugin implements Plugin<Project> {
 
     }
 
+    public abstract TaskProvider<? extends Test> registerTestTask(Project project, SourceSet sourceSet);
+
+    public abstract Class<? extends Plugin<Project>> getBasePlugin();
+
     private boolean isEnabled(ExtraPropertiesExtension extraProperties) {
         Object bwcEnabled = extraProperties.getProperties().get("bwc_tests_enabled");
         return bwcEnabled == null || (Boolean) bwcEnabled;

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/rest/compat/compat/LegacyYamlRestCompatTestPlugin.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.gradle.internal.test.rest.compat.compat;
+
+import org.elasticsearch.gradle.internal.test.rest.LegacyYamlRestTestPlugin;
+import org.elasticsearch.gradle.internal.test.rest.RestTestUtil;
+import org.gradle.api.Plugin;
+import org.gradle.api.Project;
+import org.gradle.api.file.ProjectLayout;
+import org.gradle.api.internal.file.FileOperations;
+import org.gradle.api.tasks.SourceSet;
+import org.gradle.api.tasks.TaskProvider;
+import org.gradle.api.tasks.testing.Test;
+
+import javax.inject.Inject;
+
+/**
+ * Apply this plugin to run the YAML based REST tests from a prior major version against this version's cluster.
+ *
+ * @deprecated use {@link YamlRestCompatTestPlugin}
+ */
+@Deprecated
+public class LegacyYamlRestCompatTestPlugin extends AbstractYamlRestCompatTestPlugin {
+    @Inject
+    public LegacyYamlRestCompatTestPlugin(ProjectLayout projectLayout, FileOperations fileOperations) {
+        super(projectLayout, fileOperations);
+    }
+
+    @Override
+    public TaskProvider<? extends Test> registerTestTask(Project project, SourceSet sourceSet) {
+        return RestTestUtil.registerTestTask(project, sourceSet, sourceSet.getTaskName(null, "test"));
+    }
+
+    @Override
+    public Class<? extends Plugin<Project>> getBasePlugin() {
+        return LegacyYamlRestTestPlugin.class;
+    }
+}

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/rest/compat/RestCompatTestTransformTask.java → build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/rest/compat/compat/RestCompatTestTransformTask.java

@@ -6,7 +6,7 @@
  * Side Public License, v 1.
  */
 
-package org.elasticsearch.gradle.internal.rest.compat;
+package org.elasticsearch.gradle.internal.test.rest.compat.compat;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/test/rest/compat/compat/YamlRestCompatTestPlugin.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.gradle.internal.test.rest.compat.compat;
+
+import org.elasticsearch.gradle.internal.test.rest.InternalYamlRestTestPlugin;
+import org.elasticsearch.gradle.internal.test.rest.RestTestUtil;
+import org.elasticsearch.gradle.testclusters.StandaloneRestIntegTestTask;
+import org.gradle.api.Plugin;
+import org.gradle.api.Project;
+import org.gradle.api.file.ProjectLayout;
+import org.gradle.api.internal.file.FileOperations;
+import org.gradle.api.tasks.SourceSet;
+import org.gradle.api.tasks.TaskProvider;
+import org.gradle.api.tasks.testing.Test;
+
+import javax.inject.Inject;
+
+/**
+ * Apply this plugin to run the YAML based REST tests from a prior major version against this version's cluster.
+ */
+public class YamlRestCompatTestPlugin extends AbstractYamlRestCompatTestPlugin {
+    @Inject
+    public YamlRestCompatTestPlugin(ProjectLayout projectLayout, FileOperations fileOperations) {
+        super(projectLayout, fileOperations);
+    }
+
+    @Override
+    public TaskProvider<? extends Test> registerTestTask(Project project, SourceSet sourceSet) {
+        return RestTestUtil.registerTestTask(project, sourceSet, sourceSet.getTaskName(null, "test"), StandaloneRestIntegTestTask.class);
+    }
+
+    @Override
+    public Class<? extends Plugin<Project>> getBasePlugin() {
+        return InternalYamlRestTestPlugin.class;
+    }
+}

build-tools/src/main/java/org/elasticsearch/gradle/testclusters/StandaloneRestIntegTestTask.java

@@ -73,11 +73,6 @@ public class StandaloneRestIntegTestTask extends Test implements TestClustersAwa
         this.debugServer = enabled;
     }
 
-    @Override
-    public int getMaxParallelForks() {
-        return 1;
-    }
-
     @Nested
     @Override
     public Collection<ElasticsearchCluster> getClusters() {

distribution/archives/integ-test-zip/build.gradle

@@ -8,7 +8,7 @@
 
 import org.apache.tools.ant.filters.ReplaceTokens
 
-apply plugin: 'elasticsearch.internal-java-rest-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
 
 // The integ-test-distribution is published to maven
 apply plugin: 'elasticsearch.publish'

distribution/build.gradle

@@ -116,6 +116,14 @@ def processIntegTestOutputsTaskProvider = tasks.register("processIntegTestOutput
   into integTestOutputs
 }
 
+def integTestConfigFiles = fileTree("${integTestOutputs}/config") {
+  builtBy processIntegTestOutputsTaskProvider
+}
+
+def integTestBinFiles = fileTree("${integTestOutputs}/bin") {
+  builtBy processIntegTestOutputsTaskProvider
+}
+
 def defaultModulesFiles = fileTree("${defaultOutputs}/modules") {
   builtBy processDefaultOutputsTaskProvider
 }
@@ -358,7 +366,7 @@ configure(subprojects.findAll { ['archives', 'packages'].contains(it.name) }) {
           filter("tokens" : expansionsForDistribution(distributionType, isTestDistro), ReplaceTokens.class)
         }
         from buildDefaultLog4jConfigTaskProvider
-        from defaultConfigFiles
+        from isTestDistro ? integTestConfigFiles : defaultConfigFiles
       }
     }
 
@@ -388,7 +396,7 @@ configure(subprojects.findAll { ['archives', 'packages'].contains(it.name) }) {
         // module provided bin files
         with copySpec {
           eachFile { it.setMode(0755) }
-          from(defaultBinFiles)
+          from(testDistro ? integTestBinFiles : defaultBinFiles)
           if (distributionType != 'zip') {
             exclude '*.bat'
           }

distribution/docker/build.gradle

@@ -15,7 +15,7 @@ import org.elasticsearch.gradle.util.GradleUtils
 import java.nio.file.Path
 import java.time.temporal.ChronoUnit
 
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 apply plugin: 'elasticsearch.test.fixtures'
 apply plugin: 'elasticsearch.internal-distribution-download'
 apply plugin: 'elasticsearch.dra-artifacts'

modules/aggregations/build.gradle

@@ -8,8 +8,8 @@ import org.elasticsearch.gradle.internal.info.BuildParams
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

modules/aggs-matrix-stats/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'Adds aggregations whose input are a list of numeric fields and output includes a matrix.'

modules/analysis-common/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

modules/data-streams/build.gradle

@@ -2,9 +2,9 @@ import org.elasticsearch.gradle.internal.info.BuildParams
 
 apply plugin: 'elasticsearch.test-with-dependencies'
 apply plugin: 'elasticsearch.internal-cluster-test'
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.internal-java-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'Elasticsearch Expanded Pack Plugin - Data Streams'

modules/ingest-attachment/build.gradle

@@ -7,8 +7,8 @@ import org.elasticsearch.gradle.internal.info.BuildParams
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'Ingest processor that uses Apache Tika to extract contents'

modules/ingest-common/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

modules/ingest-geoip/build.gradle

@@ -8,8 +8,8 @@
 
 import org.apache.tools.ant.taskdefs.condition.Os
 
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

modules/ingest-geoip/qa/file-based-update/build.gradle

@@ -6,7 +6,7 @@
  * Side Public License, v 1.
  */
 
-apply plugin: 'elasticsearch.internal-java-rest-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
 
 testClusters.configureEach {
   testDistribution = 'DEFAULT'

modules/ingest-user-agent/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'Ingest processor that extracts information from a user agent'

modules/kibana/build.gradle

@@ -5,7 +5,7 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-java-rest-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
 
 esplugin {
   description 'Plugin exposing APIs for Kibana system indices'

modules/lang-expression/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

modules/lang-mustache/build.gradle

@@ -5,9 +5,9 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
-apply plugin: 'elasticsearch.internal-java-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

modules/lang-painless/build.gradle

@@ -9,8 +9,8 @@
 import org.elasticsearch.gradle.testclusters.DefaultTestClustersTask;
 
 apply plugin: 'elasticsearch.validate-rest-spec'
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
     description 'An easy, safe and fast scripting language for Elasticsearch'

modules/mapper-extras/build.gradle

@@ -8,8 +8,8 @@
 
 import org.elasticsearch.gradle.internal.info.BuildParams
 
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

modules/parent-join/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

modules/percolator/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

modules/rank-eval/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

modules/reindex/build.gradle

@@ -17,9 +17,9 @@ import org.gradle.api.internal.artifacts.ArtifactAttributes
 
 apply plugin: 'elasticsearch.test-with-dependencies'
 apply plugin: 'elasticsearch.jdk-download'
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.internal-java-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

modules/repository-azure/build.gradle

@@ -13,7 +13,7 @@ import static org.elasticsearch.gradle.PropertyNormalization.IGNORE_VALUE
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 apply plugin: 'elasticsearch.internal-test-artifact-base'
 

modules/repository-gcs/build.gradle

@@ -1,7 +1,7 @@
 import org.apache.tools.ant.filters.ReplaceTokens
 import org.elasticsearch.gradle.internal.info.BuildParams
 import org.elasticsearch.gradle.internal.test.RestIntegTestTask
-import org.elasticsearch.gradle.internal.test.rest.InternalYamlRestTestPlugin
+import org.elasticsearch.gradle.internal.test.rest.LegacyYamlRestTestPlugin
 import org.elasticsearch.gradle.internal.test.InternalClusterTestPlugin
 
 import java.nio.file.Files
@@ -16,7 +16,7 @@ import static org.elasticsearch.gradle.PropertyNormalization.IGNORE_VALUE
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 apply plugin: 'elasticsearch.internal-test-artifact-base'
 
@@ -271,7 +271,7 @@ def largeBlobYamlRestTest = tasks.register("largeBlobYamlRestTest", RestIntegTes
     dependsOn "createServiceAccountFile"
   }
   SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
-  SourceSet yamlRestTestSourceSet = sourceSets.getByName(InternalYamlRestTestPlugin.SOURCE_SET_NAME)
+  SourceSet yamlRestTestSourceSet = sourceSets.getByName(LegacyYamlRestTestPlugin.SOURCE_SET_NAME)
   setTestClassesDirs(yamlRestTestSourceSet.getOutput().getClassesDirs())
   setClasspath(yamlRestTestSourceSet.getRuntimeClasspath())
 
@@ -323,7 +323,7 @@ testClusters.matching {
 if (useFixture) {
   tasks.register("yamlRestTestApplicationDefaultCredentials", RestIntegTestTask.class) {
     SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
-    SourceSet yamlRestTestSourceSet = sourceSets.getByName(InternalYamlRestTestPlugin.SOURCE_SET_NAME)
+    SourceSet yamlRestTestSourceSet = sourceSets.getByName(LegacyYamlRestTestPlugin.SOURCE_SET_NAME)
     setTestClassesDirs(yamlRestTestSourceSet.getOutput().getClassesDirs())
     setClasspath(yamlRestTestSourceSet.getRuntimeClasspath())
   }

modules/repository-s3/build.gradle

@@ -1,7 +1,7 @@
 import org.apache.tools.ant.filters.ReplaceTokens
 import org.elasticsearch.gradle.internal.info.BuildParams
 import org.elasticsearch.gradle.internal.test.RestIntegTestTask
-import org.elasticsearch.gradle.internal.test.rest.InternalYamlRestTestPlugin
+import org.elasticsearch.gradle.internal.test.rest.LegacyYamlRestTestPlugin
 import org.elasticsearch.gradle.internal.test.InternalClusterTestPlugin
 
 import static org.elasticsearch.gradle.PropertyNormalization.IGNORE_VALUE
@@ -13,7 +13,7 @@ import static org.elasticsearch.gradle.PropertyNormalization.IGNORE_VALUE
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 apply plugin: 'elasticsearch.internal-test-artifact-base'
 
@@ -244,7 +244,7 @@ if (useFixture) {
   tasks.register("yamlRestTestMinio", RestIntegTestTask) {
     description = "Runs REST tests using the Minio repository."
     SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
-    SourceSet yamlRestTestSourceSet = sourceSets.getByName(InternalYamlRestTestPlugin.SOURCE_SET_NAME)
+    SourceSet yamlRestTestSourceSet = sourceSets.getByName(LegacyYamlRestTestPlugin.SOURCE_SET_NAME)
     setTestClassesDirs(yamlRestTestSourceSet.getOutput().getClassesDirs())
     setClasspath(yamlRestTestSourceSet.getRuntimeClasspath())
 
@@ -272,7 +272,7 @@ if (useFixture) {
   tasks.register("yamlRestTestECS", RestIntegTestTask.class) {
     description = "Runs tests using the ECS repository."
     SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
-    SourceSet yamlRestTestSourceSet = sourceSets.getByName(InternalYamlRestTestPlugin.SOURCE_SET_NAME)
+    SourceSet yamlRestTestSourceSet = sourceSets.getByName(LegacyYamlRestTestPlugin.SOURCE_SET_NAME)
     setTestClassesDirs(yamlRestTestSourceSet.getOutput().getClassesDirs())
     setClasspath(yamlRestTestSourceSet.getRuntimeClasspath())
     systemProperty 'tests.rest.blacklist', [
@@ -298,7 +298,7 @@ if (useFixture) {
   tasks.register("yamlRestTestSTS", RestIntegTestTask.class) {
     description = "Runs tests with the STS (Secure Token Service)"
     SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
-    SourceSet yamlRestTestSourceSet = sourceSets.getByName(InternalYamlRestTestPlugin.SOURCE_SET_NAME)
+    SourceSet yamlRestTestSourceSet = sourceSets.getByName(LegacyYamlRestTestPlugin.SOURCE_SET_NAME)
     setTestClassesDirs(yamlRestTestSourceSet.getOutput().getClassesDirs())
     setClasspath(yamlRestTestSourceSet.getRuntimeClasspath())
     systemProperty 'tests.rest.blacklist', [

modules/repository-url/build.gradle

@@ -8,8 +8,8 @@
 
 import org.elasticsearch.gradle.PropertyNormalization
 
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 apply plugin: 'elasticsearch.test.fixtures'
 

modules/runtime-fields-common/build.gradle

@@ -7,8 +7,8 @@
  */
 
 apply plugin: 'elasticsearch.validate-rest-spec'
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'Module for runtime fields features and extensions that have large dependencies'

modules/transport-netty4/build.gradle

@@ -7,12 +7,12 @@
  */
 
 import org.elasticsearch.gradle.internal.test.RestIntegTestTask
-import org.elasticsearch.gradle.internal.test.rest.InternalJavaRestTestPlugin
+import org.elasticsearch.gradle.internal.test.rest.LegacyJavaRestTestPlugin
 import org.elasticsearch.gradle.internal.test.InternalClusterTestPlugin
 
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
-apply plugin: 'elasticsearch.internal-java-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 apply plugin: 'elasticsearch.publish'
 
@@ -73,7 +73,7 @@ TaskProvider<Test> pooledInternalClusterTest = tasks.register("pooledInternalClu
 
 TaskProvider<RestIntegTestTask> pooledJavaRestTest = tasks.register("pooledJavaRestTest", RestIntegTestTask) {
     SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
-    SourceSet javaRestTestSourceSet = sourceSets.getByName(InternalJavaRestTestPlugin.SOURCE_SET_NAME)
+    SourceSet javaRestTestSourceSet = sourceSets.getByName(LegacyJavaRestTestPlugin.SOURCE_SET_NAME)
     setTestClassesDirs(javaRestTestSourceSet.getOutput().getClassesDirs())
     setClasspath(javaRestTestSourceSet.getRuntimeClasspath())
 

plugins/analysis-icu/build.gradle

@@ -7,8 +7,8 @@ import org.elasticsearch.gradle.internal.info.BuildParams
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

plugins/analysis-kuromoji/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'The Japanese (kuromoji) Analysis plugin integrates Lucene kuromoji analysis module into elasticsearch.'

plugins/analysis-nori/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'The Korean (nori) Analysis plugin integrates Lucene nori analysis module into elasticsearch.'

plugins/analysis-phonetic/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'The Phonetic Analysis plugin integrates phonetic token filter analysis with elasticsearch.'

plugins/analysis-smartcn/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'Smart Chinese Analysis plugin integrates Lucene Smart Chinese analysis module into elasticsearch.'

plugins/analysis-stempel/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'The Stempel (Polish) Analysis plugin integrates Lucene stempel (polish) analysis module into elasticsearch.'

plugins/analysis-ukrainian/build.gradle

@@ -5,8 +5,8 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'The Ukrainian Analysis plugin integrates the Lucene UkrainianMorfologikAnalyzer into elasticsearch.'

plugins/discovery-azure-classic/build.gradle

@@ -8,7 +8,7 @@ import org.elasticsearch.gradle.internal.info.BuildParams
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

plugins/discovery-ec2/build.gradle

@@ -7,7 +7,7 @@ import org.elasticsearch.gradle.internal.info.BuildParams
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

plugins/discovery-ec2/qa/amazon-ec2/build.gradle

@@ -10,11 +10,11 @@ import org.apache.tools.ant.filters.ReplaceTokens
 import org.elasticsearch.gradle.internal.info.BuildParams
 import org.elasticsearch.gradle.internal.test.AntFixture
 import org.elasticsearch.gradle.internal.test.RestIntegTestTask
-import org.elasticsearch.gradle.internal.test.rest.InternalYamlRestTestPlugin
+import org.elasticsearch.gradle.internal.test.rest.LegacyYamlRestTestPlugin
 
 import static org.elasticsearch.gradle.PropertyNormalization.IGNORE_VALUE
 
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 
 dependencies {
   yamlRestTestImplementation project(':plugins:discovery-ec2')
@@ -62,7 +62,7 @@ tasks.named("yamlRestTest").configure { enabled = false }
   def yamlRestTestTask = tasks.register("yamlRestTest${action}", RestIntegTestTask) {
     dependsOn fixture
     SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
-    SourceSet yamlRestTestSourceSet = sourceSets.getByName(InternalYamlRestTestPlugin.SOURCE_SET_NAME)
+    SourceSet yamlRestTestSourceSet = sourceSets.getByName(LegacyYamlRestTestPlugin.SOURCE_SET_NAME)
     testClassesDirs = yamlRestTestSourceSet.getOutput().getClassesDirs()
     classpath = yamlRestTestSourceSet.getRuntimeClasspath()
   }

plugins/discovery-gce/build.gradle

@@ -1,4 +1,4 @@
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

plugins/discovery-gce/qa/gce/build.gradle

@@ -13,7 +13,7 @@ import org.elasticsearch.gradle.internal.test.AntFixture
 
 import static org.elasticsearch.gradle.PropertyNormalization.IGNORE_VALUE
 
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 
 final int gceNumberOfNodes = 3
 

plugins/mapper-annotated-text/build.gradle

@@ -7,8 +7,8 @@ import org.elasticsearch.gradle.internal.info.BuildParams
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'The Mapper Annotated_text plugin adds support for text fields with markup used to inject annotation tokens into the index.'

plugins/mapper-murmur3/build.gradle

@@ -7,8 +7,8 @@ import org.elasticsearch.gradle.internal.info.BuildParams
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
-apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-compat-test'
 
 esplugin {
   description 'The Mapper Murmur3 plugin allows to compute hashes of a field\'s values at index-time and to store them in the index.'

plugins/mapper-size/src/yamlRestTest/java/org/elasticsearch/index/mapper/size/MapperSizeClientYamlTestSuiteIT.java

@@ -11,11 +11,16 @@ package org.elasticsearch.index.mapper.size;
 import com.carrotsearch.randomizedtesting.annotations.Name;
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
 import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
 import org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase;
+import org.junit.ClassRule;
 
 public class MapperSizeClientYamlTestSuiteIT extends ESClientYamlSuiteTestCase {
 
+    @ClassRule
+    public static ElasticsearchCluster cluster = ElasticsearchCluster.local().plugin("mapper-size").build();
+
     public MapperSizeClientYamlTestSuiteIT(@Name("yaml") ClientYamlTestCandidate testCandidate) {
         super(testCandidate);
     }
@@ -24,4 +29,9 @@ public class MapperSizeClientYamlTestSuiteIT extends ESClientYamlSuiteTestCase {
     public static Iterable<Object[]> parameters() throws Exception {
         return createParameters();
     }
+
+    @Override
+    protected String getTestRestCluster() {
+        return cluster.getHttpAddresses();
+    }
 }

plugins/repository-hdfs/build.gradle

@@ -15,8 +15,8 @@ import java.nio.file.Path
 import static org.elasticsearch.gradle.PropertyNormalization.IGNORE_VALUE
 
 apply plugin: 'elasticsearch.test.fixtures'
-apply plugin: 'elasticsearch.internal-java-rest-test'
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 
 esplugin {
     description 'The HDFS repository plugin adds support for Hadoop Distributed File-System (HDFS) repositories.'

plugins/store-smb/build.gradle

@@ -5,7 +5,7 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 apply plugin: 'elasticsearch.internal-cluster-test'
 
 esplugin {

qa/ccs-unavailable-clusters/build.gradle

@@ -5,7 +5,7 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-java-rest-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
 
 testClusters.matching { it.name == "javaRestTest" }.configureEach {
   setting 'xpack.security.enabled', 'true'

qa/smoke-test-http/build.gradle

@@ -6,7 +6,7 @@
  * Side Public License, v 1.
  */
 
-apply plugin: 'elasticsearch.internal-java-rest-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
 //apply plugin: 'elasticsearch.test-with-dependencies'
 
 dependencies {

qa/smoke-test-ingest-disabled/build.gradle

@@ -5,7 +5,7 @@
  * in compliance with, at your election, the Elastic License 2.0 or the Server
  * Side Public License, v 1.
  */
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 
 dependencies {
   testImplementation project(':modules:ingest-common')

qa/smoke-test-ingest-with-all-dependencies/build.gradle

@@ -6,7 +6,7 @@
  * Side Public License, v 1.
  */
 
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 
 dependencies {
   yamlRestTestImplementation project(':modules:lang-mustache')
@@ -24,4 +24,4 @@ tasks.named("yamlRestTestTestingConventions").configure {
 
 tasks.named("forbiddenPatterns").configure {
   exclude '**/*.mmdb'
-}
+}

qa/smoke-test-multinode/build.gradle

@@ -10,7 +10,7 @@
 import org.elasticsearch.gradle.Version
 import org.elasticsearch.gradle.internal.info.BuildParams
 
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 
 restResources {
   restTests {

qa/smoke-test-plugins/build.gradle

@@ -9,7 +9,7 @@
 import org.apache.tools.ant.filters.ReplaceTokens
 import org.elasticsearch.gradle.internal.info.BuildParams
 
-apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 
 ext.pluginPaths = []
 project(':plugins').getChildProjects().each { pluginName, pluginProject ->

qa/system-indices/build.gradle

@@ -7,7 +7,7 @@
  */
 
 apply plugin: 'elasticsearch.base-internal-es-plugin'
-apply plugin: 'elasticsearch.internal-java-rest-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
 
 esplugin {
   name 'system-indices-qa'

qa/unconfigured-node-name/build.gradle

@@ -6,7 +6,7 @@
  * Side Public License, v 1.
  */
 
-apply plugin: 'elasticsearch.internal-java-rest-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
 
 testClusters.configureEach {
   setting 'xpack.security.enabled', 'false'

rest-api-spec/build.gradle

@@ -33,9 +33,8 @@ artifacts {
   restTests(new File(projectDir, "src/yamlRestTest/resources/rest-api-spec/test"))
 }
 
-testClusters.configureEach {
-  module ':modules:mapper-extras'
-  requiresFeature 'es.index_mode_feature_flag_registered', Version.fromString("8.0.0")
+dependencies {
+  clusterModules project(":modules:mapper-extras")
 }
 
 tasks.named("yamlRestTestV7CompatTransform").configure { task ->

rest-api-spec/src/yamlRestTest/java/org/elasticsearch/test/rest/ClientYamlTestSuiteIT.java

@@ -13,8 +13,11 @@ import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 import com.carrotsearch.randomizedtesting.annotations.TimeoutSuite;
 
 import org.apache.lucene.tests.util.TimeUnits;
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.cluster.FeatureFlag;
 import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
 import org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase;
+import org.junit.ClassRule;
 
 /** Rest integration test. Runs against a cluster started by {@code gradle integTest} */
 
@@ -22,6 +25,12 @@ import org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase;
 @TimeoutSuite(millis = 40 * TimeUnits.MINUTE)
 public class ClientYamlTestSuiteIT extends ESClientYamlSuiteTestCase {
 
+    @ClassRule
+    public static ElasticsearchCluster cluster = ElasticsearchCluster.local()
+        .module("mapper-extras")
+        .feature(FeatureFlag.TIME_SERIES_MODE)
+        .build();
+
     public ClientYamlTestSuiteIT(@Name("yaml") ClientYamlTestCandidate testCandidate) {
         super(testCandidate);
     }
@@ -30,4 +39,9 @@ public class ClientYamlTestSuiteIT extends ESClientYamlSuiteTestCase {
     public static Iterable<Object[]> parameters() throws Exception {
         return createParameters();
     }
+
+    @Override
+    protected String getTestRestCluster() {
+        return cluster.getHttpAddresses();
+    }
 }

settings.gradle

@@ -83,8 +83,9 @@ List projects = [
   'test:fixtures:url-fixture',
   'test:fixtures:nginx-fixture',
   'test:logger-usage',
-  'test:yaml-rest-runner',
-  'test:x-content'
+  'test:test-clusters',
+  'test:x-content',
+  'test:yaml-rest-runner'
 ]
 
 /**

test/external-modules/build.gradle

@@ -3,7 +3,7 @@ import org.elasticsearch.gradle.internal.info.BuildParams
 
 subprojects {
   apply plugin: 'elasticsearch.base-internal-es-plugin'
-  apply plugin: 'elasticsearch.internal-yaml-rest-test'
+  apply plugin: 'elasticsearch.legacy-yaml-rest-test'
 
   esplugin {
     name it.name

test/external-modules/die-with-dignity/build.gradle

@@ -1,7 +1,7 @@
 import org.elasticsearch.gradle.internal.info.BuildParams
 import org.elasticsearch.gradle.util.GradleUtils
 
-apply plugin: 'elasticsearch.internal-java-rest-test'
+apply plugin: 'elasticsearch.legacy-java-rest-test'
 apply plugin: 'elasticsearch.internal-es-plugin'
 
 esplugin {

test/test-clusters/build.gradle

@@ -0,0 +1,15 @@
+import org.elasticsearch.gradle.internal.conventions.util.Util
+
+apply plugin: 'elasticsearch.java'
+apply plugin: 'com.github.johnrengelman.shadow'
+
+dependencies {
+  shadow "junit:junit:${versions.junit}"
+  shadow "org.apache.logging.log4j:log4j-api:${versions.log4j}"
+
+  implementation "org.elasticsearch.gradle:reaper"
+}
+
+tasks.named("processResources").configure {
+  from(new File(Util.locateElasticsearchWorkspace(gradle), "build-tools-internal/version.properties"))
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/ClusterFactory.java

@@ -0,0 +1,14 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster;
+
+public interface ClusterFactory<S extends ClusterSpec, H extends ClusterHandle> {
+
+    H create(S spec);
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/ClusterHandle.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster;
+
+import java.io.Closeable;
+
+/**
+ * A handle to an {@link ElasticsearchCluster}.
+ */
+public interface ClusterHandle extends Closeable {
+    /**
+     * Starts the cluster. This method will block until all nodes are started and cluster is ready to serve requests.
+     */
+    void start();
+
+    /**
+     * Stops the cluster. This method will block until all cluster node processes have exited. This method is thread-safe and subsequent
+     * calls will wait for the exiting termination to complete.
+     *
+     * @param forcibly whether to forcibly terminate the cluster
+     */
+    void stop(boolean forcibly);
+
+    /**
+     * Whether the cluster is started or not. This method makes no guarantees on cluster availability, only that the node processes have
+     * been started.
+     *
+     * @return whether the cluster has been started
+     */
+    boolean isStarted();
+
+    /**
+     * Returns a comma-separated list of HTTP transport endpoints for cluster. If this method is called on an unstarted cluster, the cluster
+     * will be started. This method is thread-safe and subsequent calls will wait for cluster start and availability.
+     *
+     * @return cluster node HTTP transport addresses
+     */
+    String getHttpAddresses();
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/ClusterSpec.java

@@ -0,0 +1,11 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster;
+
+public interface ClusterSpec {}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/ElasticsearchCluster.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster;
+
+import org.elasticsearch.test.cluster.local.DefaultLocalClusterSpecBuilder;
+import org.elasticsearch.test.cluster.local.LocalClusterSpecBuilder;
+import org.junit.rules.TestRule;
+
+/**
+ * <p>A JUnit test rule for orchestrating an Elasticsearch cluster for local integration testing. New clusters can be created via one of the
+ * various static builder methods. For example:</p>
+ * <pre>
+ * &#064;ClassRule
+ * public static ElasticsearchCluster myCluster = ElasticsearchCluster.local().build();
+ * </pre>
+ */
+public interface ElasticsearchCluster extends TestRule, ClusterHandle {
+
+    /**
+     * Creates a new {@link DefaultLocalClusterSpecBuilder} for defining a locally orchestrated cluster. Local clusters use a locally built
+     * Elasticsearch distribution.
+     *
+     * @return a builder for a local cluster
+     */
+    static LocalClusterSpecBuilder local() {
+        return new DefaultLocalClusterSpecBuilder();
+    }
+
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/EnvironmentProvider.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster;
+
+import org.elasticsearch.test.cluster.local.LocalClusterSpec.LocalNodeSpec;
+
+import java.util.Map;
+
+/**
+ * Functional interface for supplying environment variables to an Elasticsearch node. This interface is designed to be implemented by tests
+ * and fixtures wanting to provide settings to an {@link ElasticsearchCluster} in a dynamic fashion. Instances are evaluated lazily at
+ * cluster start time.
+ */
+public interface EnvironmentProvider {
+
+    /**
+     * Returns a collection of environment variables to apply to an Elasticsearch cluster node. This method is called when the cluster is
+     * started so implementors can return dynamic environment values that may or may not be based on the given node spec.
+     *
+     * @param nodeSpec the specification for the given node to apply settings to
+     * @return environment variables to add to the node
+     */
+    Map<String, String> get(LocalNodeSpec nodeSpec);
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/FeatureFlag.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster;
+
+import org.elasticsearch.test.cluster.util.Version;
+
+/**
+ * Elasticsearch feature flags. Used in conjunction with {@link org.elasticsearch.test.cluster.local.LocalSpecBuilder#feature(FeatureFlag)}
+ * to indicate that this feature is required and should be enabled when appropriate.
+ */
+public enum FeatureFlag {
+    TIME_SERIES_MODE("es.index_mode_feature_flag_registered=true", Version.fromString("8.0.0"), null);
+
+    public final String systemProperty;
+    public final Version from;
+    public final Version until;
+
+    FeatureFlag(String systemProperty, Version from, Version until) {
+        this.systemProperty = systemProperty;
+        this.from = from;
+        this.until = until;
+    }
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/MutableSettingsProvider.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster;
+
+import org.elasticsearch.test.cluster.local.LocalClusterSpec;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class MutableSettingsProvider implements SettingsProvider {
+    private final Map<String, String> settings = new HashMap<>();
+
+    @Override
+    public Map<String, String> get(LocalClusterSpec.LocalNodeSpec nodeSpec) {
+        return settings;
+    }
+
+    public void put(String setting, String value) {
+        settings.put(setting, value);
+    }
+
+    public void clear() {
+        settings.clear();
+    }
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/SettingsProvider.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster;
+
+import org.elasticsearch.test.cluster.local.LocalClusterSpec.LocalNodeSpec;
+
+import java.util.Map;
+
+/**
+ * Functional interface for supplying settings to an Elasticsearch node. This interface is designed to be implemented by tests and fixtures
+ * wanting to provide settings to an {@link ElasticsearchCluster} in a dynamic fashion. Instances are evaluated lazily at cluster
+ * start time.
+ */
+public interface SettingsProvider {
+
+    /**
+     * Returns a collection of settings to apply to an Elasticsearch cluster node. This method is called when the cluster is started so
+     * implementors can return dynamic setting values that may or may not be based on the given node spec.
+     *
+     * @param nodeSpec the specification for the given node to apply settings to
+     * @return settings to add to the node
+     */
+    Map<String, String> get(LocalNodeSpec nodeSpec);
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/AbstractLocalSpecBuilder.java

@@ -0,0 +1,159 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster.local;
+
+import org.elasticsearch.test.cluster.EnvironmentProvider;
+import org.elasticsearch.test.cluster.FeatureFlag;
+import org.elasticsearch.test.cluster.SettingsProvider;
+import org.elasticsearch.test.cluster.local.distribution.DistributionType;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Supplier;
+
+public abstract class AbstractLocalSpecBuilder<T extends LocalSpecBuilder<?>> implements LocalSpecBuilder<T> {
+    private final AbstractLocalSpecBuilder<?> parent;
+    private final List<SettingsProvider> settingsProviders = new ArrayList<>();
+    private final Map<String, String> settings = new HashMap<>();
+    private final List<EnvironmentProvider> environmentProviders = new ArrayList<>();
+    private final Map<String, String> environment = new HashMap<>();
+    private final Set<String> modules = new HashSet<>();
+    private final Set<String> plugins = new HashSet<>();
+    private final Set<FeatureFlag> features = new HashSet<>();
+    private DistributionType distributionType;
+
+    protected AbstractLocalSpecBuilder(AbstractLocalSpecBuilder<?> parent) {
+        this.parent = parent;
+    }
+
+    @Override
+    public T settings(SettingsProvider settingsProvider) {
+        this.settingsProviders.add(settingsProvider);
+        return cast(this);
+    }
+
+    List<SettingsProvider> getSettingsProviders() {
+        return inherit(() -> parent.getSettingsProviders(), settingsProviders);
+    }
+
+    @Override
+    public T setting(String setting, String value) {
+        this.settings.put(setting, value);
+        return cast(this);
+    }
+
+    Map<String, String> getSettings() {
+        return inherit(() -> parent.getSettings(), settings);
+    }
+
+    @Override
+    public T environment(EnvironmentProvider environmentProvider) {
+        this.environmentProviders.add(environmentProvider);
+        return cast(this);
+    }
+
+    List<EnvironmentProvider> getEnvironmentProviders() {
+        return inherit(() -> parent.getEnvironmentProviders(), environmentProviders);
+
+    }
+
+    @Override
+    public T environment(String key, String value) {
+        this.environment.put(key, value);
+        return cast(this);
+    }
+
+    Map<String, String> getEnvironment() {
+        return inherit(() -> parent.getEnvironment(), environment);
+    }
+
+    @Override
+    public T distribution(DistributionType type) {
+        this.distributionType = type;
+        return cast(this);
+    }
+
+    DistributionType getDistributionType() {
+        return inherit(() -> parent.getDistributionType(), distributionType);
+    }
+
+    @Override
+    public T module(String moduleName) {
+        this.modules.add(moduleName);
+        return cast(this);
+    }
+
+    Set<String> getModules() {
+        return inherit(() -> parent.getModules(), modules);
+    }
+
+    @Override
+    public T plugin(String pluginName) {
+        this.plugins.add(pluginName);
+        return cast(this);
+    }
+
+    Set<String> getPlugins() {
+        return inherit(() -> parent.getPlugins(), plugins);
+    }
+
+    @Override
+    public T feature(FeatureFlag feature) {
+        this.features.add(feature);
+        return cast(this);
+    }
+
+    Set<FeatureFlag> getFeatures() {
+        return inherit(() -> parent.getFeatures(), features);
+    }
+
+    private <T> List<T> inherit(Supplier<List<T>> parent, List<T> child) {
+        List<T> combinedList = new ArrayList<>();
+        if (this.parent != null) {
+            combinedList.addAll(parent.get());
+        }
+        combinedList.addAll(child);
+        return combinedList;
+    }
+
+    private <T> Set<T> inherit(Supplier<Set<T>> parent, Set<T> child) {
+        Set<T> combinedSet = new HashSet<>();
+        if (this.parent != null) {
+            combinedSet.addAll(parent.get());
+        }
+        combinedSet.addAll(child);
+        return combinedSet;
+    }
+
+    private <K, V> Map<K, V> inherit(Supplier<Map<K, V>> parent, Map<K, V> child) {
+        Map<K, V> combinedMap = new HashMap<>();
+        if (this.parent != null) {
+            combinedMap.putAll(parent.get());
+        }
+        combinedMap.putAll(child);
+        return combinedMap;
+    }
+
+    private <T> T inherit(Supplier<T> parent, T child) {
+        T value = null;
+        if (this.parent != null) {
+            value = parent.get();
+        }
+        return child == null ? value : child;
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> T cast(Object o) {
+        return (T) o;
+    }
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/DefaultEnvironmentProvider.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster.local;
+
+import org.elasticsearch.test.cluster.EnvironmentProvider;
+import org.elasticsearch.test.cluster.local.LocalClusterSpec.LocalNodeSpec;
+import org.elasticsearch.test.cluster.local.distribution.DistributionType;
+import org.elasticsearch.test.cluster.util.Version;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class DefaultEnvironmentProvider implements EnvironmentProvider {
+    private static final String HOSTNAME_OVERRIDE = "LinuxDarwinHostname";
+    private static final String COMPUTERNAME_OVERRIDE = "WindowsComputername";
+    private static final String TESTS_RUNTIME_JAVA_SYSPROP = "tests.runtime.java";
+
+    @Override
+    public Map<String, String> get(LocalNodeSpec nodeSpec) {
+        Map<String, String> environment = new HashMap<>();
+
+        // If we are testing the current version of Elasticsearch, use the configured runtime Java, otherwise use the bundled JDK
+        if (nodeSpec.getDistributionType() == DistributionType.INTEG_TEST || nodeSpec.getVersion().equals(Version.CURRENT)) {
+            environment.put("ES_JAVA_HOME", System.getProperty(TESTS_RUNTIME_JAVA_SYSPROP));
+        }
+
+        // Override the system hostname variables for testing
+        environment.put("HOSTNAME", HOSTNAME_OVERRIDE);
+        environment.put("COMPUTERNAME", COMPUTERNAME_OVERRIDE);
+
+        return environment;
+    }
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/DefaultLocalClusterSpecBuilder.java

@@ -0,0 +1,152 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster.local;
+
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.cluster.local.LocalClusterSpec.LocalNodeSpec;
+import org.elasticsearch.test.cluster.local.distribution.DistributionType;
+import org.elasticsearch.test.cluster.local.model.User;
+import org.elasticsearch.test.cluster.util.Version;
+import org.elasticsearch.test.cluster.util.resource.TextResource;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Consumer;
+
+public class DefaultLocalClusterSpecBuilder extends AbstractLocalSpecBuilder<LocalClusterSpecBuilder> implements LocalClusterSpecBuilder {
+    private String name = "test-cluster";
+    private final List<DefaultLocalNodeSpecBuilder> nodeBuilders = new ArrayList<>();
+    private final List<User> users = new ArrayList<>();
+    private final List<TextResource> roleFiles = new ArrayList<>();
+
+    public DefaultLocalClusterSpecBuilder() {
+        super(null);
+        this.settings(new DefaultSettingsProvider());
+        this.environment(new DefaultEnvironmentProvider());
+        this.rolesFile(TextResource.fromClasspath("default_test_roles.yml"));
+    }
+
+    @Override
+    public DefaultLocalClusterSpecBuilder name(String name) {
+        this.name = name;
+        return this;
+    }
+
+    @Override
+    public DefaultLocalClusterSpecBuilder apply(LocalClusterConfigProvider configProvider) {
+        configProvider.apply(this);
+        return this;
+    }
+
+    @Override
+    public DefaultLocalClusterSpecBuilder nodes(int nodes) {
+        if (nodes < nodeBuilders.size()) {
+            throw new IllegalArgumentException(
+                "Cannot shrink cluster to " + nodes + ". " + nodeBuilders.size() + " nodes already configured"
+            );
+        }
+
+        int newNodes = nodes - nodeBuilders.size();
+        for (int i = 0; i < newNodes; i++) {
+            nodeBuilders.add(new DefaultLocalNodeSpecBuilder(this));
+        }
+
+        return this;
+    }
+
+    @Override
+    public DefaultLocalClusterSpecBuilder withNode(Consumer<? super LocalNodeSpecBuilder> config) {
+        DefaultLocalNodeSpecBuilder builder = new DefaultLocalNodeSpecBuilder(this);
+        config.accept(builder);
+        nodeBuilders.add(builder);
+        return this;
+    }
+
+    @Override
+    public DefaultLocalClusterSpecBuilder node(int index, Consumer<? super LocalNodeSpecBuilder> config) {
+        try {
+            DefaultLocalNodeSpecBuilder builder = nodeBuilders.get(index);
+            config.accept(builder);
+        } catch (IndexOutOfBoundsException e) {
+            throw new IllegalArgumentException(
+                "No node at index + " + index + " exists. Only " + nodeBuilders.size() + " nodes have been configured"
+            );
+        }
+        return this;
+    }
+
+    @Override
+    public DefaultLocalClusterSpecBuilder user(String username, String password) {
+        this.users.add(new User(username, password));
+        return this;
+    }
+
+    @Override
+    public DefaultLocalClusterSpecBuilder user(String username, String password, String role) {
+        this.users.add(new User(username, password, role));
+        return this;
+    }
+
+    @Override
+    public DefaultLocalClusterSpecBuilder rolesFile(TextResource rolesFile) {
+        this.roleFiles.add(rolesFile);
+        return this;
+    }
+
+    @Override
+    public ElasticsearchCluster build() {
+        List<User> clusterUsers = users.isEmpty() ? List.of(User.DEFAULT_USER) : users;
+        LocalClusterSpec clusterSpec = new LocalClusterSpec(name, clusterUsers, roleFiles);
+        List<LocalNodeSpec> nodeSpecs;
+
+        if (nodeBuilders.isEmpty()) {
+            // No node-specific configuration so assume a single-node cluster
+            nodeSpecs = List.of(new DefaultLocalNodeSpecBuilder(this).build(clusterSpec));
+        } else {
+            nodeSpecs = nodeBuilders.stream().map(node -> node.build(clusterSpec)).toList();
+        }
+
+        clusterSpec.setNodes(nodeSpecs);
+        clusterSpec.validate();
+
+        return new LocalElasticsearchCluster(clusterSpec);
+    }
+
+    public static class DefaultLocalNodeSpecBuilder extends AbstractLocalSpecBuilder<LocalNodeSpecBuilder> implements LocalNodeSpecBuilder {
+        private String name;
+
+        protected DefaultLocalNodeSpecBuilder(AbstractLocalSpecBuilder<?> parent) {
+            super(parent);
+        }
+
+        @Override
+        public DefaultLocalNodeSpecBuilder name(String name) {
+            this.name = name;
+            return this;
+        }
+
+        private LocalNodeSpec build(LocalClusterSpec cluster) {
+
+            return new LocalNodeSpec(
+                cluster,
+                name,
+                Version.CURRENT,
+                getSettingsProviders(),
+                getSettings(),
+                getEnvironmentProviders(),
+                getEnvironment(),
+                getModules(),
+                getPlugins(),
+                Optional.ofNullable(getDistributionType()).orElse(DistributionType.INTEG_TEST),
+                getFeatures()
+            );
+        }
+    }
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/DefaultSettingsProvider.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster.local;
+
+import org.elasticsearch.test.cluster.SettingsProvider;
+import org.elasticsearch.test.cluster.local.LocalClusterSpec.LocalNodeSpec;
+import org.elasticsearch.test.cluster.local.distribution.DistributionType;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class DefaultSettingsProvider implements SettingsProvider {
+    @Override
+    public Map<String, String> get(LocalNodeSpec nodeSpec) {
+        Map<String, String> settings = new HashMap<>();
+
+        settings.put("node.name", nodeSpec.getName());
+        settings.put("node.attr.testattr", "test");
+        settings.put("node.portsfile", "true");
+        settings.put("http.port", "0");
+        settings.put("transport.port", "0");
+
+        if (nodeSpec.getDistributionType() == DistributionType.INTEG_TEST) {
+            settings.put("xpack.security.enabled", "false");
+        } else {
+            // Disable deprecation indexing which is enabled by default in 7.16
+            if (nodeSpec.getVersion().onOrAfter("7.16.0")) {
+                settings.put("cluster.deprecation_indexing.enabled", "false");
+            }
+        }
+
+        // Default the watermarks to absurdly low to prevent the tests from failing on nodes without enough disk space
+        settings.put("cluster.routing.allocation.disk.watermark.low", "1b");
+        settings.put("cluster.routing.allocation.disk.watermark.high", "1b");
+        settings.put("cluster.routing.allocation.disk.watermark.flood_stage", "1b");
+
+        // increase script compilation limit since tests can rapid-fire script compilations
+        if (nodeSpec.getVersion().onOrAfter("7.9.0")) {
+            settings.put("script.disable_max_compilations_rate", "true");
+        } else {
+            settings.put("script.max_compilations_rate", "2048/1m");
+        }
+
+        // Temporarily disable the real memory usage circuit breaker. It depends on real memory usage which we have no full control
+        // over and the REST client will not retry on circuit breaking exceptions yet (see #31986 for details). Once the REST client
+        // can retry on circuit breaking exceptions, we can revert again to the default configuration.
+        settings.put("indices.breaker.total.use_real_memory", "false");
+
+        // Don't wait for state, just start up quickly. This will also allow new and old nodes in the BWC case to become the master
+        settings.put("discovery.initial_state_timeout", "0s");
+
+        if (nodeSpec.getVersion().getMajor() >= 8) {
+            settings.put("cluster.service.slow_task_logging_threshold", "5s");
+            settings.put("cluster.service.slow_master_task_logging_threshold", "5s");
+        }
+
+        settings.put("action.destructive_requires_name", "false");
+
+        // Setup cluster discovery
+        String nodeNames = nodeSpec.getCluster().getNodes().stream().map(LocalNodeSpec::getName).collect(Collectors.joining(","));
+        settings.put("cluster.initial_master_nodes", "[" + nodeNames + "]");
+        settings.put("discovery.seed_providers", "file");
+        settings.put("discovery.seed_hosts", "[]");
+
+        return settings;
+    }
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/LocalClusterConfigProvider.java

@@ -0,0 +1,14 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster.local;
+
+public interface LocalClusterConfigProvider {
+
+    void apply(LocalClusterSpecBuilder builder);
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/LocalClusterFactory.java

@@ -0,0 +1,503 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster.local;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.test.cluster.ClusterFactory;
+import org.elasticsearch.test.cluster.local.LocalClusterSpec.LocalNodeSpec;
+import org.elasticsearch.test.cluster.local.distribution.DistributionDescriptor;
+import org.elasticsearch.test.cluster.local.distribution.DistributionResolver;
+import org.elasticsearch.test.cluster.local.distribution.DistributionType;
+import org.elasticsearch.test.cluster.local.model.User;
+import org.elasticsearch.test.cluster.util.IOUtils;
+import org.elasticsearch.test.cluster.util.OS;
+import org.elasticsearch.test.cluster.util.Pair;
+import org.elasticsearch.test.cluster.util.ProcessReaper;
+import org.elasticsearch.test.cluster.util.ProcessUtils;
+import org.elasticsearch.test.cluster.util.Retry;
+import org.elasticsearch.test.cluster.util.Version;
+
+import java.io.BufferedInputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class LocalClusterFactory implements ClusterFactory<LocalClusterSpec, LocalClusterHandle> {
+    private static final Logger LOGGER = LogManager.getLogger(LocalClusterFactory.class);
+    private static final Duration NODE_UP_TIMEOUT = Duration.ofMinutes(2);
+    private static final Map<Pair<Version, DistributionType>, DistributionDescriptor> TEST_DISTRIBUTIONS = new ConcurrentHashMap<>();
+    private static final String TESTS_CLUSTER_MODULES_PATH_SYSPROP = "tests.cluster.modules.path";
+    private static final String TESTS_CLUSTER_PLUGINS_PATH_SYSPROP = "tests.cluster.plugins.path";
+
+    private final Path baseWorkingDir;
+    private final DistributionResolver distributionResolver;
+
+    public LocalClusterFactory(Path baseWorkingDir, DistributionResolver distributionResolver) {
+        this.baseWorkingDir = baseWorkingDir;
+        this.distributionResolver = distributionResolver;
+    }
+
+    @Override
+    public LocalClusterHandle create(LocalClusterSpec spec) {
+        return new LocalClusterHandle(spec.getName(), spec.getNodes().stream().map(Node::new).toList());
+    }
+
+    public class Node {
+        private final LocalNodeSpec spec;
+        private final Path workingDir;
+        private final Path distributionDir;
+        private final Path snapshotsDir;
+        private final Path dataDir;
+        private final Path logsDir;
+        private final Path configDir;
+        private final Path tempDir;
+
+        private boolean initialized = false;
+        private Process process = null;
+        private DistributionDescriptor distributionDescriptor;
+
+        public Node(LocalNodeSpec spec) {
+            this.spec = spec;
+            this.workingDir = baseWorkingDir.resolve(spec.getCluster().getName()).resolve(spec.getName());
+            this.distributionDir = workingDir.resolve("distro"); // location of es distribution files, typically hard-linked
+            this.snapshotsDir = workingDir.resolve("repo");
+            this.dataDir = workingDir.resolve("data");
+            this.logsDir = workingDir.resolve("logs");
+            this.configDir = workingDir.resolve("config");
+            this.tempDir = workingDir.resolve("tmp"); // elasticsearch temporary directory
+        }
+
+        public synchronized void start() {
+            LOGGER.info("Starting Elasticsearch node '{}'", spec.getName());
+
+            if (initialized == false) {
+                LOGGER.info("Creating installation for node '{}' in {}", spec.getName(), workingDir);
+                distributionDescriptor = resolveDistribution();
+                LOGGER.info("Distribution for node '{}': {}", spec.getName(), distributionDescriptor);
+                initializeWorkingDirectory();
+                writeConfiguration();
+                createKeystore();
+                configureSecurity();
+                installPlugins();
+                if (spec.getDistributionType() == DistributionType.INTEG_TEST) {
+                    installModules();
+                }
+                initialized = true;
+            }
+
+            startElasticsearch();
+        }
+
+        public synchronized void stop(boolean forcibly) {
+            if (process != null) {
+                ProcessUtils.stopHandle(process.toHandle(), forcibly);
+                ProcessReaper.instance().unregister(getServiceName());
+            }
+        }
+
+        public void waitForExit() {
+            if (process != null) {
+                ProcessUtils.waitForExit(process.toHandle());
+            }
+        }
+
+        public String getHttpAddress() {
+            Path portFile = workingDir.resolve("logs").resolve("http.ports");
+            if (Files.notExists(portFile)) {
+                waitUntilReady();
+            }
+            return readPortsFile(portFile).get(0);
+        }
+
+        public String getTransportEndpoint() {
+            Path portsFile = workingDir.resolve("logs").resolve("transport.ports");
+            if (Files.notExists(portsFile)) {
+                waitUntilReady();
+            }
+            return readPortsFile(portsFile).get(0);
+        }
+
+        public LocalNodeSpec getSpec() {
+            return spec;
+        }
+
+        Path getWorkingDir() {
+            return workingDir;
+        }
+
+        public void waitUntilReady() {
+            try {
+                Retry.retryUntilTrue(NODE_UP_TIMEOUT, Duration.ofMillis(500), () -> {
+                    if (process.isAlive() == false) {
+                        throw new RuntimeException(
+                            "Elasticsearch process died while waiting for ports file. See console output for details."
+                        );
+                    }
+
+                    Path httpPorts = workingDir.resolve("logs").resolve("http.ports");
+                    Path transportPorts = workingDir.resolve("logs").resolve("transport.ports");
+
+                    return Files.exists(httpPorts) && Files.exists(transportPorts);
+                });
+            } catch (TimeoutException e) {
+                throw new RuntimeException("Timed out after " + NODE_UP_TIMEOUT + " waiting for ports files for: " + this);
+            } catch (ExecutionException e) {
+                throw new RuntimeException("An error occurred while waiting for ports file for: " + this, e);
+            }
+        }
+
+        private List<String> readPortsFile(Path file) {
+            try (Stream<String> lines = Files.lines(file, StandardCharsets.UTF_8)) {
+                return lines.map(String::trim).collect(Collectors.toList());
+            } catch (IOException e) {
+                throw new UncheckedIOException("Unable to read ports file: " + file, e);
+            }
+        }
+
+        private void initializeWorkingDirectory() {
+            try {
+                IOUtils.deleteWithRetry(workingDir);
+                try {
+                    IOUtils.syncWithLinks(distributionDescriptor.getDistributionDir(), distributionDir);
+                } catch (IOUtils.LinkCreationException e) {
+                    // Note does not work for network drives, e.g. Vagrant
+                    LOGGER.info("Failed to create working dir using hard links. Falling back to copy", e);
+                    // ensure we get a clean copy
+                    IOUtils.deleteWithRetry(distributionDir);
+                    IOUtils.syncWithCopy(distributionDescriptor.getDistributionDir(), distributionDir);
+                }
+                Files.createDirectories(configDir);
+                Files.createDirectories(snapshotsDir);
+                Files.createDirectories(dataDir);
+                Files.createDirectories(logsDir);
+                Files.createDirectories(tempDir);
+            } catch (IOException e) {
+                throw new UncheckedIOException("Failed to create working directory for node '" + spec.getName() + "'", e);
+            }
+        }
+
+        private DistributionDescriptor resolveDistribution() {
+            return TEST_DISTRIBUTIONS.computeIfAbsent(
+                Pair.of(spec.getVersion(), spec.getDistributionType()),
+                key -> distributionResolver.resolve(key.left, key.right)
+            );
+        }
+
+        private void writeConfiguration() {
+            Path configFile = configDir.resolve("elasticsearch.yml");
+            Path jvmOptionsFile = configDir.resolve("jvm.options");
+
+            try {
+                // Write settings to elasticsearch.yml
+                Map<String, String> pathSettings = new HashMap<>();
+                pathSettings.put("path.repo", workingDir.resolve("repo").toString());
+                pathSettings.put("path.data", workingDir.resolve("data").toString());
+                pathSettings.put("path.logs", workingDir.resolve("logs").toString());
+
+                Files.writeString(
+                    configFile,
+                    Stream.concat(spec.resolveSettings().entrySet().stream(), pathSettings.entrySet().stream())
+                        .map(entry -> entry.getKey() + ": " + entry.getValue())
+                        .collect(Collectors.joining("\n")),
+                    StandardOpenOption.TRUNCATE_EXISTING,
+                    StandardOpenOption.CREATE
+                );
+
+                // Copy additional configuration from distribution
+                try (Stream<Path> configFiles = Files.list(distributionDir.resolve("config"))) {
+                    for (Path file : configFiles.toList()) {
+                        Path dest = configFile.getParent().resolve(file.getFileName());
+                        if (Files.exists(dest) == false) {
+                            Files.copy(file, dest);
+                        }
+                    }
+                }
+
+                // Patch jvm.options file to update paths
+                String content = Files.readString(jvmOptionsFile);
+                Map<String, String> expansions = getJvmOptionsReplacements();
+                for (String key : expansions.keySet()) {
+                    if (content.contains(key) == false) {
+                        throw new IOException("Template property '" + key + "' not found in template.");
+                    }
+                    content = content.replace(key, expansions.get(key));
+                }
+                Files.writeString(jvmOptionsFile, content);
+            } catch (IOException e) {
+                throw new UncheckedIOException("Could not write config file: " + configFile, e);
+            }
+        }
+
+        private void createKeystore() {
+            try {
+                ProcessUtils.exec(
+                    workingDir,
+                    OS.conditional(
+                        c -> c.onWindows(() -> distributionDir.resolve("bin").resolve("elasticsearch-keystore.bat"))
+                            .onUnix(() -> distributionDir.resolve("bin").resolve("elasticsearch-keystore"))
+                    ),
+                    getEnvironmentVariables(),
+                    false,
+                    "-v",
+                    "create"
+                ).waitFor();
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        }
+
+        private void configureSecurity() {
+            if (spec.isSecurityEnabled()) {
+                if (spec.getUsers().isEmpty() == false) {
+                    LOGGER.info("Setting up roles.yml for node '{}'", spec.getName());
+
+                    Path destination = workingDir.resolve("config").resolve("roles.yml");
+                    spec.getRolesFiles().forEach(rolesFile -> {
+                        try {
+                            Files.writeString(
+                                destination,
+                                rolesFile.getText() + System.lineSeparator(),
+                                StandardCharsets.UTF_8,
+                                StandardOpenOption.APPEND
+                            );
+                        } catch (IOException e) {
+                            throw new UncheckedIOException("Failed to append roles file " + rolesFile + " to " + destination, e);
+                        }
+                    });
+                }
+
+                LOGGER.info("Creating users for node '{}'", spec.getName());
+                for (User user : spec.getUsers()) {
+                    try {
+                        ProcessUtils.exec(
+                            workingDir,
+                            distributionDir.resolve("bin").resolve("elasticsearch-users"),
+                            getEnvironmentVariables(),
+                            false,
+                            "useradd",
+                            user.getUsername(),
+                            "-p",
+                            user.getPassword(),
+                            "-r",
+                            user.getRole()
+                        ).waitFor();
+                    } catch (InterruptedException e) {
+                        throw new RuntimeException(e);
+                    }
+                }
+            }
+        }
+
+        private void installPlugins() {
+            if (spec.getPlugins().isEmpty() == false) {
+                Pattern pattern = Pattern.compile("(.+)(?:-\\d\\.\\d\\.\\d-SNAPSHOT\\.zip)?");
+
+                LOGGER.info("Installing plugins {} into node '{}", spec.getPlugins(), spec.getName());
+                List<Path> pluginPaths = Arrays.stream(System.getProperty(TESTS_CLUSTER_PLUGINS_PATH_SYSPROP).split(File.pathSeparator))
+                    .map(Path::of)
+                    .toList();
+
+                List<String> toInstall = spec.getPlugins()
+                    .stream()
+                    .map(
+                        pluginName -> pluginPaths.stream()
+                            .map(path -> Pair.of(pattern.matcher(path.getFileName().toString()), path))
+                            .filter(pair -> pair.left.matches())
+                            .map(p -> p.right.getParent().resolve(p.left.group(1)))
+                            .findFirst()
+                            .orElseThrow(() -> {
+                                String taskPath = System.getProperty("tests.task");
+                                String project = taskPath.substring(0, taskPath.lastIndexOf(':'));
+
+                                throw new RuntimeException(
+                                    "Unable to locate plugin '"
+                                        + pluginName
+                                        + "'. Ensure you've added the following to the build script for project '"
+                                        + project
+                                        + "':\n\n"
+                                        + "dependencies {\n"
+                                        + "  clusterModules "
+                                        + "project(':plugins:"
+                                        + pluginName
+                                        + "')"
+                                        + "\n}"
+                                );
+                            })
+                    )
+                    .map(p -> p.toUri().toString())
+                    .toList();
+
+                Path pluginCommand = OS.conditional(
+                    c -> c.onWindows(() -> distributionDir.resolve("bin").resolve("elasticsearch-plugin.bat"))
+                        .onUnix(() -> distributionDir.resolve("bin").resolve("elasticsearch-plugin"))
+                );
+                if (spec.getVersion().onOrAfter("7.6.0")) {
+                    try {
+                        ProcessUtils.exec(
+                            workingDir,
+                            pluginCommand,
+                            getEnvironmentVariables(),
+                            false,
+                            Stream.concat(Stream.of("install", "--batch"), toInstall.stream()).toArray(String[]::new)
+                        ).waitFor();
+                    } catch (InterruptedException e) {
+                        throw new RuntimeException(e);
+                    }
+                } else {
+                    toInstall.forEach(plugin -> {
+                        try {
+                            ProcessUtils.exec(workingDir, pluginCommand, getEnvironmentVariables(), false, "install", "--batch", plugin)
+                                .waitFor();
+                        } catch (InterruptedException e) {
+                            throw new RuntimeException(e);
+                        }
+                    });
+                }
+            }
+        }
+
+        private void installModules() {
+            if (spec.getModules().isEmpty() == false) {
+                LOGGER.info("Installing modules {} into node '{}", spec.getModules(), spec.getName());
+                List<Path> modulePaths = Arrays.stream(System.getProperty(TESTS_CLUSTER_MODULES_PATH_SYSPROP).split(File.pathSeparator))
+                    .map(Path::of)
+                    .toList();
+
+                spec.getModules().forEach(module -> installModule(module, modulePaths));
+            }
+        }
+
+        private void installModule(String moduleName, List<Path> modulePaths) {
+            Path destination = distributionDir.resolve("modules").resolve(moduleName);
+            if (Files.notExists(destination)) {
+                Path modulePath = modulePaths.stream().filter(path -> path.endsWith(moduleName)).findFirst().orElseThrow(() -> {
+                    String taskPath = System.getProperty("tests.task");
+                    String project = taskPath.substring(0, taskPath.lastIndexOf(':'));
+                    String moduleDependency = moduleName.startsWith("x-pack")
+                        ? "project(xpackModule('" + moduleName.substring(7) + "'))"
+                        : "project(':modules:" + moduleName + "')";
+
+                    throw new RuntimeException(
+                        "Unable to locate module '"
+                            + moduleName
+                            + "'. Ensure you've added the following to the build script for project '"
+                            + project
+                            + "':\n\n"
+                            + "dependencies {\n"
+                            + "  clusterModules "
+                            + moduleDependency
+                            + "\n}"
+                    );
+
+                });
+
+                IOUtils.syncWithCopy(modulePath.getParent(), destination);
+
+                // Install any extended plugins
+                Properties pluginProperties = new Properties();
+                try (
+                    InputStream in = new BufferedInputStream(
+                        new FileInputStream(modulePath.resolve("plugin-descriptor.properties").toFile())
+                    )
+                ) {
+                    pluginProperties.load(in);
+                    String extendedProperty = pluginProperties.getProperty("extended.plugins");
+                    if (extendedProperty != null) {
+                        String[] extendedPlugins = extendedProperty.split(",");
+                        for (String plugin : extendedPlugins) {
+                            installModule(plugin, modulePaths);
+                        }
+                    }
+                } catch (IOException e) {
+                    throw new UncheckedIOException(e);
+                }
+            }
+        }
+
+        private void startElasticsearch() {
+            process = ProcessUtils.exec(
+                workingDir,
+                OS.conditional(
+                    c -> c.onWindows(() -> distributionDir.resolve("bin").resolve("elasticsearch.bat"))
+                        .onUnix(() -> distributionDir.resolve("bin").resolve("elasticsearch"))
+                ),
+                getEnvironmentVariables(),
+                true
+            );
+
+            ProcessReaper.instance().registerPid(getServiceName(), process.pid());
+        }
+
+        private Map<String, String> getEnvironmentVariables() {
+            Map<String, String> environment = new HashMap<>(spec.resolveEnvironment());
+            environment.put("ES_PATH_CONF", workingDir.resolve("config").toString());
+            environment.put("ES_TMPDIR", workingDir.resolve("tmp").toString());
+            // Windows requires this as it defaults to `c:\windows` despite ES_TMPDIR
+            environment.put("TMP", workingDir.resolve("tmp").toString());
+
+            String featureFlagProperties = "";
+            if (spec.getFeatures().isEmpty() == false && distributionDescriptor.isSnapshot() == false) {
+                featureFlagProperties = spec.getFeatures()
+                    .stream()
+                    .filter(f -> spec.getVersion().onOrAfter(f.from) && (f.until == null || spec.getVersion().before(f.until)))
+                    .map(f -> "-D" + f.systemProperty)
+                    .collect(Collectors.joining(" "));
+            }
+
+            String heapSize = System.getProperty("tests.heap.size", "512m");
+            environment.put("ES_JAVA_OPTS", "-Xms" + heapSize + " -Xmx" + heapSize + " -ea -esa "
+            // Support passing in additional JVM arguments
+                + System.getProperty("tests.jvm.argline", "")
+                + " "
+                + featureFlagProperties);
+
+            return environment;
+        }
+
+        private Map<String, String> getJvmOptionsReplacements() {
+            Path relativeLogsDir = workingDir.relativize(logsDir);
+            return Map.of(
+                "-XX:HeapDumpPath=data",
+                "-XX:HeapDumpPath=" + relativeLogsDir,
+                "logs/gc.log",
+                relativeLogsDir.resolve("gc.log").toString(),
+                "-XX:ErrorFile=logs/hs_err_pid%p.log",
+                "-XX:ErrorFile=" + relativeLogsDir.resolve("hs_err_pid%p.log")
+            );
+        }
+
+        private String getServiceName() {
+            return baseWorkingDir.getFileName() + "-" + spec.getCluster().getName() + "-" + spec.getName();
+        }
+
+        @Override
+        public String toString() {
+            return "{ cluster: '" + spec.getCluster().getName() + "', node: '" + spec.getName() + "' }";
+        }
+    }
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/LocalClusterHandle.java

@@ -0,0 +1,155 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster.local;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.test.cluster.ClusterHandle;
+import org.elasticsearch.test.cluster.local.LocalClusterFactory.Node;
+import org.elasticsearch.test.cluster.local.model.User;
+import org.elasticsearch.test.cluster.util.ExceptionUtils;
+import org.elasticsearch.test.cluster.util.Retry;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinWorkerThread;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+
+public class LocalClusterHandle implements ClusterHandle {
+    private static final Logger LOGGER = LogManager.getLogger(LocalClusterHandle.class);
+    private static final Duration CLUSTER_UP_TIMEOUT = Duration.ofSeconds(30);
+
+    public final ForkJoinPool executor = new ForkJoinPool(
+        Math.max(Runtime.getRuntime().availableProcessors(), 4),
+        new ForkJoinPool.ForkJoinWorkerThreadFactory() {
+            private final AtomicLong counter = new AtomicLong(0);
+
+            @Override
+            public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
+                ForkJoinWorkerThread thread = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
+                thread.setName(name + "-node-executor-" + counter.getAndIncrement());
+                return thread;
+            }
+        },
+        null,
+        false
+    );
+    private final AtomicBoolean started = new AtomicBoolean(false);
+    private final String name;
+    private final List<Node> nodes;
+
+    public LocalClusterHandle(String name, List<Node> nodes) {
+        this.name = name;
+        this.nodes = nodes;
+    }
+
+    @Override
+    public void start() {
+        if (started.getAndSet(true) == false) {
+            LOGGER.info("Starting Elasticsearch test cluster '{}'", name);
+            execute(() -> nodes.parallelStream().forEach(Node::start));
+        }
+        waitUntilReady();
+    }
+
+    @Override
+    public void stop(boolean forcibly) {
+        if (started.getAndSet(false)) {
+            LOGGER.info("Stopping Elasticsearch test cluster '{}', forcibly: {}", name, forcibly);
+            execute(() -> nodes.forEach(n -> n.stop(forcibly)));
+        } else {
+            // Make sure the process is stopped, otherwise wait
+            execute(() -> nodes.forEach(n -> n.waitForExit()));
+        }
+    }
+
+    @Override
+    public boolean isStarted() {
+        return started.get();
+    }
+
+    @Override
+    public void close() {
+        stop(false);
+
+        executor.shutdownNow();
+        try {
+            executor.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public String getHttpAddresses() {
+        start();
+        return execute(() -> nodes.parallelStream().map(Node::getHttpAddress).collect(Collectors.joining(",")));
+    }
+
+    private void waitUntilReady() {
+        writeUnicastHostsFile();
+        try {
+            Retry.retryUntilTrue(CLUSTER_UP_TIMEOUT, Duration.ZERO, () -> {
+                Node node = nodes.get(0);
+                String scheme = node.getSpec().isSettingTrue("xpack.security.http.ssl.enabled") ? "https" : "http";
+                WaitForHttpResource wait = new WaitForHttpResource(scheme, node.getHttpAddress(), nodes.size());
+                User credentials = node.getSpec().getUsers().get(0);
+                wait.setUsername(credentials.getUsername());
+                wait.setPassword(credentials.getPassword());
+                return wait.wait(500);
+            });
+        } catch (TimeoutException e) {
+            throw new RuntimeException("Timed out after " + CLUSTER_UP_TIMEOUT + " waiting for cluster '" + name + "' status to be yellow");
+        } catch (ExecutionException e) {
+            throw new RuntimeException("An error occurred while checking cluster '" + name + "' status.", e);
+        }
+    }
+
+    private void writeUnicastHostsFile() {
+        String transportUris = execute(() -> nodes.parallelStream().map(Node::getTransportEndpoint).collect(Collectors.joining("\n")));
+        nodes.forEach(node -> {
+            try {
+                Path hostsFile = node.getWorkingDir().resolve("config").resolve("unicast_hosts.txt");
+                if (Files.notExists(hostsFile)) {
+                    Files.writeString(hostsFile, transportUris);
+                }
+            } catch (IOException e) {
+                throw new UncheckedIOException("Failed to write unicast_hosts for: " + node, e);
+            }
+        });
+    }
+
+    private <T> T execute(Callable<T> task) {
+        try {
+            return executor.submit(task).get();
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        } catch (ExecutionException e) {
+            throw new RuntimeException("An error occurred orchestrating test cluster.", ExceptionUtils.findRootCause(e));
+        }
+    }
+
+    private void execute(Runnable task) {
+        execute(() -> {
+            task.run();
+            return true;
+        });
+    }
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/LocalClusterSpec.java

@@ -0,0 +1,190 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster.local;
+
+import org.elasticsearch.test.cluster.ClusterSpec;
+import org.elasticsearch.test.cluster.EnvironmentProvider;
+import org.elasticsearch.test.cluster.FeatureFlag;
+import org.elasticsearch.test.cluster.SettingsProvider;
+import org.elasticsearch.test.cluster.local.distribution.DistributionType;
+import org.elasticsearch.test.cluster.local.model.User;
+import org.elasticsearch.test.cluster.util.Version;
+import org.elasticsearch.test.cluster.util.resource.TextResource;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class LocalClusterSpec implements ClusterSpec {
+    private final String name;
+    private final List<User> users;
+    private final List<TextResource> roleFiles;
+    private List<LocalNodeSpec> nodes;
+
+    public LocalClusterSpec(String name, List<User> users, List<TextResource> roleFiles) {
+        this.name = name;
+        this.users = users;
+        this.roleFiles = roleFiles;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public List<User> getUsers() {
+        return users;
+    }
+
+    public List<TextResource> getRoleFiles() {
+        return roleFiles;
+    }
+
+    public List<LocalNodeSpec> getNodes() {
+        return nodes;
+    }
+
+    public void setNodes(List<LocalNodeSpec> nodes) {
+        this.nodes = nodes;
+    }
+
+    void validate() {
+        // Ensure we don't have nodes with duplicate names
+        List<String> nodeNames = nodes.stream().map(LocalNodeSpec::getName).collect(Collectors.toList());
+        Set<String> uniqueNames = nodes.stream().map(LocalNodeSpec::getName).collect(Collectors.toSet());
+        uniqueNames.forEach(name -> nodeNames.remove(nodeNames.indexOf(name)));
+
+        if (nodeNames.isEmpty() == false) {
+            throw new IllegalArgumentException("Cluster cannot contain nodes with duplicates names: " + nodeNames);
+        }
+    }
+
+    public static class LocalNodeSpec {
+        private final LocalClusterSpec cluster;
+        private final String name;
+        private final Version version;
+        private final List<SettingsProvider> settingsProviders;
+        private final Map<String, String> settings;
+        private final List<EnvironmentProvider> environmentProviders;
+        private final Map<String, String> environment;
+        private final Set<String> modules;
+        private final Set<String> plugins;
+        private final DistributionType distributionType;
+        private final Set<FeatureFlag> features;
+
+        public LocalNodeSpec(
+            LocalClusterSpec cluster,
+            String name,
+            Version version,
+            List<SettingsProvider> settingsProviders,
+            Map<String, String> settings,
+            List<EnvironmentProvider> environmentProviders,
+            Map<String, String> environment,
+            Set<String> modules,
+            Set<String> plugins,
+            DistributionType distributionType,
+            Set<FeatureFlag> features
+        ) {
+            this.cluster = cluster;
+            this.name = name;
+            this.version = version;
+            this.settingsProviders = settingsProviders;
+            this.settings = settings;
+            this.environmentProviders = environmentProviders;
+            this.environment = environment;
+            this.modules = modules;
+            this.plugins = plugins;
+            this.distributionType = distributionType;
+            this.features = features;
+        }
+
+        public LocalClusterSpec getCluster() {
+            return cluster;
+        }
+
+        public String getName() {
+            return name == null ? cluster.getName() + "-" + cluster.getNodes().indexOf(this) : name;
+        }
+
+        public Version getVersion() {
+            return version;
+        }
+
+        public List<User> getUsers() {
+            return cluster.getUsers();
+        }
+
+        public List<TextResource> getRolesFiles() {
+            return cluster.getRoleFiles();
+        }
+
+        public DistributionType getDistributionType() {
+            return distributionType;
+        }
+
+        public Set<String> getModules() {
+            return modules;
+        }
+
+        public Set<String> getPlugins() {
+            return plugins;
+        }
+
+        public Set<FeatureFlag> getFeatures() {
+            return features;
+        }
+
+        public boolean isSecurityEnabled() {
+            return Boolean.parseBoolean(
+                resolveSettings().getOrDefault("xpack.security.enabled", getVersion().onOrAfter("8.0.0") ? "true" : "false")
+            );
+        }
+
+        public boolean isSettingTrue(String setting) {
+            return Boolean.parseBoolean(resolveSettings().getOrDefault(setting, "false"));
+        }
+
+        /**
+         * Resolve node settings. Order of precedence is as follows:
+         * <ol>
+         *     <li>Settings from cluster configured {@link SettingsProvider}</li>
+         *     <li>Settings from node configured {@link SettingsProvider}</li>
+         *     <li>Explicit cluster settings</li>
+         *     <li>Explicit node settings</li>
+         * </ol>
+         *
+         * @return resolved settings for node
+         */
+        public Map<String, String> resolveSettings() {
+            Map<String, String> resolvedSettings = new HashMap<>();
+            settingsProviders.forEach(p -> resolvedSettings.putAll(p.get(this)));
+            resolvedSettings.putAll(settings);
+            return resolvedSettings;
+        }
+
+        /**
+         * Resolve node environment variables. Order of precedence is as follows:
+         * <ol>
+         *     <li>Environment variables from cluster configured {@link EnvironmentProvider}</li>
+         *     <li>Environment variables from node configured {@link EnvironmentProvider}</li>
+         *     <li>Environment variables cluster settings</li>
+         *     <li>Environment variables node settings</li>
+         * </ol>
+         *
+         * @return resolved environment variables for node
+         */
+        public Map<String, String> resolveEnvironment() {
+            Map<String, String> resolvedEnvironment = new HashMap<>();
+            environmentProviders.forEach(p -> resolvedEnvironment.putAll(p.get(this)));
+            resolvedEnvironment.putAll(environment);
+            return resolvedEnvironment;
+        }
+    }
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/LocalClusterSpecBuilder.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster.local;
+
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.cluster.util.resource.TextResource;
+
+import java.util.function.Consumer;
+
+public interface LocalClusterSpecBuilder extends LocalSpecBuilder<LocalClusterSpecBuilder> {
+    /**
+     * Sets the node name. By default, "test-cluster" is used.
+     */
+    LocalClusterSpecBuilder name(String name);
+
+    LocalClusterSpecBuilder apply(LocalClusterConfigProvider configProvider);
+
+    /**
+     * Sets the number of nodes for the cluster.
+     */
+    LocalClusterSpecBuilder nodes(int nodes);
+
+    /**
+     * Adds a new node to the cluster and configures the node.
+     */
+    LocalClusterSpecBuilder withNode(Consumer<? super LocalNodeSpecBuilder> config);
+
+    /**
+     * Configures an existing node.
+     *
+     * @param index the index of the node to configure
+     * @param config configuration to apply to the node
+     */
+    LocalClusterSpecBuilder node(int index, Consumer<? super LocalNodeSpecBuilder> config);
+
+    /**
+     * Register a user using the default test role.
+     */
+    LocalClusterSpecBuilder user(String username, String password);
+
+    /**
+     * Register a user using the given role.
+     */
+    LocalClusterSpecBuilder user(String username, String password, String role);
+
+    /**
+     * Register a roles file with cluster via the supplied {@link TextResource}.
+     */
+    LocalClusterSpecBuilder rolesFile(TextResource rolesFile);
+
+    ElasticsearchCluster build();
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/LocalElasticsearchCluster.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster.local;
+
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.cluster.local.distribution.LocalDistributionResolver;
+import org.elasticsearch.test.cluster.local.distribution.SnapshotDistributionResolver;
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+
+import java.nio.file.Path;
+
+public class LocalElasticsearchCluster implements ElasticsearchCluster {
+    private final LocalClusterSpec spec;
+    private LocalClusterHandle handle;
+
+    public LocalElasticsearchCluster(LocalClusterSpec spec) {
+        this.spec = spec;
+    }
+
+    @Override
+    public Statement apply(Statement base, Description description) {
+        return new Statement() {
+            @Override
+            public void evaluate() throws Throwable {
+                try {
+                    handle = new LocalClusterFactory(
+                        Path.of(System.getProperty("java.io.tmpdir")).resolve(description.getDisplayName()).toAbsolutePath(),
+                        new LocalDistributionResolver(new SnapshotDistributionResolver())
+                    ).create(spec);
+                    handle.start();
+                    base.evaluate();
+                } finally {
+                    close();
+                }
+            }
+        };
+    }
+
+    @Override
+    public void start() {
+        checkHandle();
+        handle.start();
+    }
+
+    @Override
+    public void stop(boolean forcibly) {
+        checkHandle();
+        handle.stop(forcibly);
+    }
+
+    @Override
+    public boolean isStarted() {
+        checkHandle();
+        return handle.isStarted();
+    }
+
+    @Override
+    public void close() {
+        checkHandle();
+        handle.close();
+    }
+
+    @Override
+    public String getHttpAddresses() {
+        checkHandle();
+        return handle.getHttpAddresses();
+    }
+
+    private void checkHandle() {
+        if (handle == null) {
+            throw new IllegalStateException("Cluster handle has not been initialized. Did you forget the @ClassRule annotation?");
+        }
+    }
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/LocalNodeSpecBuilder.java

@@ -0,0 +1,17 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster.local;
+
+public interface LocalNodeSpecBuilder extends LocalSpecBuilder<LocalNodeSpecBuilder> {
+
+    /**
+     * Sets the node name. By default, nodes are named after the cluster with an incrementing suffix (ex: my-cluster-0).
+     */
+    LocalNodeSpecBuilder name(String name);
+}

test/test-clusters/src/main/java/org/elasticsearch/test/cluster/local/LocalSpecBuilder.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.cluster.local;
+
+import org.elasticsearch.test.cluster.EnvironmentProvider;
+import org.elasticsearch.test.cluster.FeatureFlag;
+import org.elasticsearch.test.cluster.SettingsProvider;
+import org.elasticsearch.test.cluster.local.distribution.DistributionType;
+
+interface LocalSpecBuilder<T extends LocalSpecBuilder<?>> {
+    /**
+     * Register a {@link SettingsProvider}.
+     */
+    T settings(SettingsProvider settingsProvider);
+
+    /**
+     * Add a new node setting.
+     */
+    T setting(String setting, String value);
+
+    /**
+     * Register a {@link EnvironmentProvider}.
+     */
+    T environment(EnvironmentProvider environmentProvider);
+
+    /**
+     * Add a new node environment variable.
+     */
+    T environment(String key, String value);
+
+    /**
+     * Set the cluster {@link DistributionType}. By default, the {@link DistributionType#INTEG_TEST} distribution is used.
+     */
+    T distribution(DistributionType type);
+
+    /**
+     * Ensure module is installed into the distribution when using the {@link DistributionType#INTEG_TEST} distribution. This is ignored
+     * when the {@link DistributionType#DEFAULT} is being used.
+     */
+    T module(String moduleName);
+
+    /**
+     * Ensure plugin is installed into the distribution.
+     */
+    T plugin(String pluginName);
+
+    /**
+     * Require feature to be enabled in the cluster.
+     */
+    T feature(FeatureFlag feature);
+}