Fix and unmute CrossClusterEsqlRCS2UnavailableRemotesIT tests (elastic#120402)

* Fix and unmute `CrossClusterEsqlRCS2UnavailableRemotesIT` tests

* [CI] Auto commit changes from spotless

* Fix test

---------

Co-authored-by: elasticsearchmachine <[email protected]>

Author: pawankartik-elastic

muted-tests.yml

@@ -131,9 +131,6 @@ tests:
 - class: org.elasticsearch.xpack.ml.integration.RegressionIT
   method: testTwoJobsWithSameRandomizeSeedUseSameTrainingSet
   issue: https://github.com/elastic/elasticsearch/issues/117805
-- class: org.elasticsearch.xpack.remotecluster.CrossClusterEsqlRCS2UnavailableRemotesIT
-  method: testEsqlRcs2UnavailableRemoteScenarios
-  issue: https://github.com/elastic/elasticsearch/issues/117419
 - class: org.elasticsearch.xpack.esql.action.EsqlActionTaskIT
   method: testCancelRequestWhenFailingFetchingPages
   issue: https://github.com/elastic/elasticsearch/issues/118193

x-pack/plugin/security/qa/multi-cluster/src/javaRestTest/java/org/elasticsearch/xpack/remotecluster/CrossClusterEsqlRCS2UnavailableRemotesIT.java

@@ -25,10 +25,12 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.oneOf;
 
 public class CrossClusterEsqlRCS2UnavailableRemotesIT extends AbstractRemoteClusterSecurityTestCase {
     private static final AtomicReference<Map<String, Object>> API_KEY_MAP_REF = new AtomicReference<>();
@@ -179,8 +181,10 @@ private void remoteClusterShutdownWithSkipUnavailableTrue() throws Exception {
             Map<String, ?> failuresMap = (Map<String, ?>) remoteClusterFailures.get(0);
 
             Map<String, ?> reason = (Map<String, ?>) failuresMap.get("reason");
-            assertThat(reason.get("type").toString(), equalTo("connect_transport_exception"));
-            assertThat(reason.get("reason").toString(), containsString("Unable to connect to [my_remote_cluster]"));
+            assertThat(
+                reason.get("type").toString(),
+                oneOf("node_disconnected_exception", "connect_transport_exception", "node_not_connected_exception")
+            );
         } finally {
             fulfillingCluster.start();
             closeFulfillingClusterClient();
@@ -201,7 +205,14 @@ private void remoteClusterShutdownWithSkipUnavailableFalse() throws Exception {
             // A simple query that targets our remote cluster.
             String query = "FROM *,my_remote_cluster:* | LIMIT 10";
             ResponseException ex = expectThrows(ResponseException.class, () -> performRequestWithRemoteSearchUser(esqlRequest(query)));
-            assertThat(ex.getMessage(), containsString("connect_transport_exception"));
+            assertThat(
+                ex.getMessage(),
+                anyOf(
+                    containsString("node_disconnected_exception"),
+                    containsString("connect_transport_exception"),
+                    containsString("node_not_connected_exception")
+                )
+            );
         } finally {
             fulfillingCluster.start();
             closeFulfillingClusterClient();