Re-enable shard rebalance in ES|QL rest tests (elastic#122889) (elastic#122897)

Today, we disable shard rebalancing in ES|QL REST tests to avoid 
failures when shards are rebalanced while the query is being executed.
This is no longer necessary if the clusters are on version 8.19 or 9.1.

Author: dnhatn

x-pack/plugin/esql/qa/server/mixed-cluster/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/mixed/Clusters.java

@@ -14,15 +14,22 @@
 public class Clusters {
     public static ElasticsearchCluster mixedVersionCluster() {
         Version oldVersion = Version.fromString(System.getProperty("tests.old_cluster_version"));
-        return ElasticsearchCluster.local()
+        var cluster = ElasticsearchCluster.local()
             .distribution(DistributionType.DEFAULT)
             .withNode(node -> node.version(oldVersion))
             .withNode(node -> node.version(Version.CURRENT))
             .withNode(node -> node.version(oldVersion))
             .withNode(node -> node.version(Version.CURRENT))
             .setting("xpack.security.enabled", "false")
-            .setting("xpack.license.self_generated.type", "trial")
-            .setting("cluster.routing.rebalance.enable", "none") // disable relocation until we have retry in ESQL
-            .build();
+            .setting("xpack.license.self_generated.type", "trial");
+        if (supportRetryOnShardFailures(oldVersion) == false) {
+            cluster.setting("cluster.routing.rebalance.enable", "none");
+        }
+        return cluster.build();
+    }
+
+    private static boolean supportRetryOnShardFailures(Version version) {
+        return version.onOrAfter(Version.fromString("9.1.0"))
+            || (version.onOrAfter(Version.fromString("8.19.0")) && version.before(Version.fromString("9.0.0")));
     }
 }

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/Clusters.java

@@ -17,38 +17,44 @@ public class Clusters {
     static final String LOCAL_CLUSTER_NAME = "local_cluster";
 
     public static ElasticsearchCluster remoteCluster() {
-        return ElasticsearchCluster.local()
+        Version version = distributionVersion("tests.version.remote_cluster");
+        var cluster = ElasticsearchCluster.local()
             .name(REMOTE_CLUSTER_NAME)
             .distribution(DistributionType.DEFAULT)
-            .version(distributionVersion("tests.version.remote_cluster"))
+            .version(version)
             .nodes(2)
             .setting("node.roles", "[data,ingest,master]")
             .setting("xpack.security.enabled", "false")
             .setting("xpack.license.self_generated.type", "trial")
-            .shared(true)
-            .setting("cluster.routing.rebalance.enable", "none")
-            .build();
+            .shared(true);
+        if (supportRetryOnShardFailures(version) == false) {
+            cluster.setting("cluster.routing.rebalance.enable", "none");
+        }
+        return cluster.build();
     }
 
     public static ElasticsearchCluster localCluster(ElasticsearchCluster remoteCluster) {
         return localCluster(remoteCluster, true);
     }
 
     public static ElasticsearchCluster localCluster(ElasticsearchCluster remoteCluster, Boolean skipUnavailable) {
-        return ElasticsearchCluster.local()
+        Version version = distributionVersion("tests.version.local_cluster");
+        var cluster = ElasticsearchCluster.local()
             .name(LOCAL_CLUSTER_NAME)
             .distribution(DistributionType.DEFAULT)
-            .version(distributionVersion("tests.version.local_cluster"))
+            .version(version)
             .nodes(2)
             .setting("xpack.security.enabled", "false")
             .setting("xpack.license.self_generated.type", "trial")
             .setting("node.roles", "[data,ingest,master,remote_cluster_client]")
             .setting("cluster.remote.remote_cluster.seeds", () -> "\"" + remoteCluster.getTransportEndpoint(0) + "\"")
             .setting("cluster.remote.connections_per_cluster", "1")
             .setting("cluster.remote." + REMOTE_CLUSTER_NAME + ".skip_unavailable", skipUnavailable.toString())
-            .shared(true)
-            .setting("cluster.routing.rebalance.enable", "none")
-            .build();
+            .shared(true);
+        if (supportRetryOnShardFailures(version) == false) {
+            cluster.setting("cluster.routing.rebalance.enable", "none");
+        }
+        return cluster.build();
     }
 
     public static org.elasticsearch.Version localClusterVersion() {
@@ -65,4 +71,9 @@ private static Version distributionVersion(String key) {
         final String val = System.getProperty(key);
         return val != null ? Version.fromString(val) : Version.CURRENT;
     }
+
+    private static boolean supportRetryOnShardFailures(Version version) {
+        return version.onOrAfter(Version.fromString("9.1.0"))
+            || (version.onOrAfter(Version.fromString("8.19.0")) && version.before(Version.fromString("9.0.0")));
+    }
 }