Fix flaky test caused by unreliable remote cluster stalling

Author: pawankartik-elastic

muted-tests.yml

@@ -479,9 +479,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.ccq.MultiClustersIT
   method: testLookupJoinAliasesSkipOld
   issue: https://github.com/elastic/elasticsearch/issues/131697
-- class: org.elasticsearch.indices.cluster.RemoteSearchForceConnectTimeoutIT
-  method: testTimeoutSetting
-  issue: https://github.com/elastic/elasticsearch/issues/131656
 - class: org.elasticsearch.xpack.stack.StackYamlIT
   method: test {yaml=stack/10_basic/Test wrong data_stream type - logs from 9.2.0}
   issue: https://github.com/elastic/elasticsearch/issues/131803
@@ -605,9 +602,6 @@ tests:
 - class: org.elasticsearch.index.mapper.vectors.DenseVectorFieldIndexTypeUpdateIT
   method: testDenseVectorMappingUpdate {initialType=int4_hnsw updateType=bbq_disk}
   issue: https://github.com/elastic/elasticsearch/issues/132165
-- class: org.elasticsearch.indices.cluster.FieldCapsForceConnectTimeoutIT
-  method: testTimeoutSetting
-  issue: https://github.com/elastic/elasticsearch/issues/132179
 - class: org.elasticsearch.index.mapper.vectors.DenseVectorFieldIndexTypeUpdateIT
   method: "testDenseVectorMappingUpdate {initialType=bbq_flat updateType=bbq_disk #2}"
   issue: https://github.com/elastic/elasticsearch/issues/132184

server/src/internalClusterTest/java/org/elasticsearch/indices/cluster/FieldCapsForceConnectTimeoutIT.java

@@ -73,21 +73,23 @@ protected Map<String, Boolean> skipUnavailableForRemoteClusters() {
 
     public void testTimeoutSetting() {
         var latch = new CountDownLatch(1);
-        for (String nodeName : cluster(LOCAL_CLUSTER).getNodeNames()) {
-            MockTransportService mts = (MockTransportService) cluster(LOCAL_CLUSTER).getInstance(TransportService.class, nodeName);
-
-            mts.addConnectBehavior(
-                cluster(LINKED_CLUSTER_1).getInstance(TransportService.class, (String) null),
-                ((transport, discoveryNode, profile, listener) -> {
+        for (String localNodeName : cluster(LOCAL_CLUSTER).getNodeNames()) {
+            MockTransportService localMts = (MockTransportService) cluster(LOCAL_CLUSTER).getInstance(
+                TransportService.class,
+                localNodeName
+            );
+            for (String remoteNodeName : cluster(LINKED_CLUSTER_1).getNodeNames()) {
+                TransportService remoteTs = cluster(LINKED_CLUSTER_1).getInstance(TransportService.class, remoteNodeName);
+                localMts.addConnectBehavior(remoteTs, ((transport, discoveryNode, profile, listener) -> {
                     try {
                         latch.await();
                     } catch (InterruptedException e) {
                         throw new AssertionError(e);
                     }
 
                     transport.openConnection(discoveryNode, profile, listener);
-                })
-            );
+                }));
+            }
         }
 
         // Add some dummy data to prove we are communicating fine with the remote.

server/src/internalClusterTest/java/org/elasticsearch/indices/cluster/RemoteSearchForceConnectTimeoutIT.java

@@ -70,21 +70,23 @@ protected Map<String, Boolean> skipUnavailableForRemoteClusters() {
 
     public void testTimeoutSetting() {
         var latch = new CountDownLatch(1);
-        for (String nodeName : cluster(LOCAL_CLUSTER).getNodeNames()) {
-            MockTransportService mts = (MockTransportService) cluster(LOCAL_CLUSTER).getInstance(TransportService.class, nodeName);
-
-            mts.addConnectBehavior(
-                cluster(REMOTE_CLUSTER_1).getInstance(TransportService.class, (String) null),
-                ((transport, discoveryNode, profile, listener) -> {
+        for (String localNodeName : cluster(LOCAL_CLUSTER).getNodeNames()) {
+            MockTransportService localMts = (MockTransportService) cluster(LOCAL_CLUSTER).getInstance(
+                TransportService.class,
+                localNodeName
+            );
+            for (String remoteNodeName : cluster(REMOTE_CLUSTER_1).getNodeNames()) {
+                TransportService remoteTs = cluster(REMOTE_CLUSTER_1).getInstance(TransportService.class, remoteNodeName);
+                localMts.addConnectBehavior(remoteTs, ((transport, discoveryNode, profile, listener) -> {
                     try {
                         latch.await();
                     } catch (InterruptedException e) {
                         throw new AssertionError(e);
                     }
 
                     transport.openConnection(discoveryNode, profile, listener);
-                })
-            );
+                }));
+            }
         }
 
         // Add some dummy data to prove we are communicating fine with the remote.