Force Connect Test: Fix flaky test caused by unreliable remote cluster stalling (#132641)

Fix flaky test caused by unreliable remote cluster stalling

Author: pawankartik-elastic

muted-tests.yml

@@ -455,9 +455,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.packaging.test.DockerTests
   method: test151MachineDependentHeapWithSizeOverride
   issue: https://github.com/elastic/elasticsearch/issues/123437
@@ -473,9 +470,6 @@ tests:
 - class: org.elasticsearch.index.engine.MergeWithLowDiskSpaceIT
   method: testRelocationWhileForceMerging
   issue: https://github.com/elastic/elasticsearch/issues/131789
-- class: org.elasticsearch.indices.cluster.FieldCapsForceConnectTimeoutIT
-  method: testTimeoutSetting
-  issue: https://github.com/elastic/elasticsearch/issues/132179
 - class: org.elasticsearch.test.rest.yaml.MDPYamlTestSuiteIT
   method: test {yaml=mdp/10_basic/Index using shared data path}
   issue: https://github.com/elastic/elasticsearch/issues/132223

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.