Wait for connect on remote settings update (#48497)

This is related to #47718. It introduces a 10 seconds wait for a
connection to complete when remote clsuter settings introduce a new
remote cluster connection.

Author: Tim-Brooks

server/src/main/java/org/elasticsearch/transport/RemoteClusterService.java

@@ -49,6 +49,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.BiFunction;
@@ -396,7 +397,24 @@ protected void updateRemoteCluster(String clusterAlias, List<String> addresses,
             builder.setPingInterval(pingSchedule);
             newProfile = builder.build();
         }
-        updateRemoteCluster(clusterAlias, addresses, proxyAddress, newProfile, noopListener);
+
+        if (remoteClusters.containsKey(clusterAlias) == false) {
+            CountDownLatch latch = new CountDownLatch(1);
+            updateRemoteCluster(clusterAlias, addresses, proxyAddress, newProfile, ActionListener.wrap(latch::countDown));
+
+            try {
+                // Wait 10 seconds for a new cluster. We must use a latch instead of a future because we
+                // are on the cluster state thread and our custom future implementation will throw an
+                // assertion.
+                if (latch.await(10, TimeUnit.SECONDS) == false) {
+                    logger.warn("failed to connect to new remote cluster {} within {}", clusterAlias, TimeValue.timeValueSeconds(10));
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        } else {
+            updateRemoteCluster(clusterAlias, addresses, proxyAddress, newProfile, noopListener);
+        }
     }
 
     void updateRemoteCluster(final String clusterAlias, final List<String> addresses, final String proxyAddress,

server/src/test/java/org/elasticsearch/transport/RemoteClusterServiceTests.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.OriginalIndices;
 import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
 import org.elasticsearch.common.Strings;
@@ -322,7 +323,19 @@ public void testIncrementallyAddClusters() throws IOException {
                     assertFalse(service.isCrossClusterSearchEnabled());
                     service.initializeRemoteClusters();
                     assertFalse(service.isCrossClusterSearchEnabled());
-                    service.updateRemoteCluster("cluster_1", Collections.singletonList(cluster1Seed.getAddress().toString()), null);
+
+                    PlainActionFuture<Void> clusterAdded = PlainActionFuture.newFuture();
+                    // Add the cluster on a different thread to test that we wait for a new cluster to
+                    // connect before returning.
+                    new Thread(() -> {
+                        try {
+                            service.updateRemoteCluster("cluster_1", Collections.singletonList(cluster1Seed.getAddress().toString()), null);
+                            clusterAdded.onResponse(null);
+                        } catch (Exception e) {
+                            clusterAdded.onFailure(e);
+                        }
+                    }).start();
+                    clusterAdded.actionGet();
                     assertTrue(service.isCrossClusterSearchEnabled());
                     assertTrue(service.isRemoteClusterRegistered("cluster_1"));
                     service.updateRemoteCluster("cluster_2", Collections.singletonList(cluster2Seed.getAddress().toString()), null);

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrSingleNodeTestCase.java

@@ -67,11 +67,9 @@ public void setupLocalRemote() throws Exception {
         updateSettingsRequest.transientSettings(Settings.builder().put("cluster.remote.local.seeds", address));
         assertAcked(client().admin().cluster().updateSettings(updateSettingsRequest).actionGet());
 
-        assertBusy(() -> {
-            List<RemoteConnectionInfo> infos = client().execute(RemoteInfoAction.INSTANCE, new RemoteInfoRequest()).get().getInfos();
-            assertThat(infos.size(), equalTo(1));
-            assertThat(infos.get(0).getNumNodesConnected(), equalTo(1));
-        });
+        List<RemoteConnectionInfo> infos = client().execute(RemoteInfoAction.INSTANCE, new RemoteInfoRequest()).get().getInfos();
+        assertThat(infos.size(), equalTo(1));
+        assertThat(infos.get(0).getNumNodesConnected(), equalTo(1));
     }
 
     @Before

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/RestartIndexFollowingIT.java

@@ -94,12 +94,10 @@ private void setupRemoteCluster() throws Exception {
         updateSettingsRequest.persistentSettings(Settings.builder().put("cluster.remote.leader_cluster.seeds", address));
         assertAcked(followerClient().admin().cluster().updateSettings(updateSettingsRequest).actionGet());
 
-        assertBusy(() -> {
-            List<RemoteConnectionInfo> infos =
-                followerClient().execute(RemoteInfoAction.INSTANCE, new RemoteInfoRequest()).get().getInfos();
-            assertThat(infos.size(), equalTo(1));
-            assertThat(infos.get(0).getNumNodesConnected(), greaterThanOrEqualTo(1));
-        });
+        List<RemoteConnectionInfo> infos =
+            followerClient().execute(RemoteInfoAction.INSTANCE, new RemoteInfoRequest()).get().getInfos();
+        assertThat(infos.size(), equalTo(1));
+        assertThat(infos.get(0).getNumNodesConnected(), greaterThanOrEqualTo(1));
     }
 
     private void cleanRemoteCluster() throws Exception {