Flaky RemoteClustersIT: Add assert busy to avoid race condition (open…

…search-project#11057) Signed-off-by: Ankit Kala <[email protected]>
bowenlan-amzn · Nov 8, 2023 · 4d8c228 · 4d8c228
1 parent 0ba5d58
commit 4d8c228
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 9 deletions.
diff --git a/qa/remote-clusters/src/test/java/org/opensearch/cluster/remote/test/RemoteClustersIT.java b/qa/remote-clusters/src/test/java/org/opensearch/cluster/remote/test/RemoteClustersIT.java
@@ -42,11 +42,13 @@
 import org.opensearch.client.cluster.RemoteInfoRequest;
 import org.opensearch.client.indices.CreateIndexRequest;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.xcontent.XContentFactory;
 import org.junit.After;
 import org.junit.Before;
 
 import java.io.IOException;
+import java.util.concurrent.TimeUnit;
 
 
 public class RemoteClustersIT extends AbstractMultiClusterRemoteTestCase {
@@ -112,7 +114,7 @@ public void testSniffModeConnectionFails() throws IOException {
         assertFalse(rci.isConnected());
     }
 
-    public void testHAProxyModeConnectionWorks() throws IOException {
+    public void testHAProxyModeConnectionWorks() throws Exception {
         String proxyAddress = "haproxy:9600";
         logger.info("Configuring remote cluster [{}]", proxyAddress);
         ClusterUpdateSettingsRequest request = new ClusterUpdateSettingsRequest().persistentSettings(Settings.builder()
@@ -121,12 +123,14 @@ public void testHAProxyModeConnectionWorks() throws IOException {
             .build());
         assertTrue(cluster1Client().cluster().putSettings(request, RequestOptions.DEFAULT).isAcknowledged());
 
-        RemoteConnectionInfo rci = cluster1Client().cluster().remoteInfo(new RemoteInfoRequest(), RequestOptions.DEFAULT).getInfos().get(0);
-        logger.info("Connection info: {}", rci);
-        if (!rci.isConnected()) {    
-            logger.info("Cluster health: {}", cluster1Client().cluster().health(new ClusterHealthRequest(), RequestOptions.DEFAULT));
-        }
-        assertTrue(rci.isConnected());
+        assertBusy(() -> {
+            RemoteConnectionInfo rci = cluster1Client().cluster().remoteInfo(new RemoteInfoRequest(), RequestOptions.DEFAULT).getInfos().get(0);
+            logger.info("Connection info: {}", rci);
+            if (!rci.isConnected()) {
+                logger.info("Cluster health: {}", cluster1Client().cluster().health(new ClusterHealthRequest(), RequestOptions.DEFAULT));
+            }
+            assertTrue(rci.isConnected());
+        }, 10, TimeUnit.SECONDS);
 
         assertEquals(2L, cluster1Client().search(
             new SearchRequest("haproxynosn:test2"), RequestOptions.DEFAULT).getHits().getTotalHits().value);

diff --git a/server/src/main/java/org/opensearch/transport/ProxyConnectionStrategy.java b/server/src/main/java/org/opensearch/transport/ProxyConnectionStrategy.java
@@ -309,7 +309,7 @@ public void onResponse(Void v) {
 
                     @Override
                     public void onFailure(Exception e) {
-                        logger.debug(
+                        logger.error(
                             new ParameterizedMessage(
                                 "failed to open remote connection [remote cluster: {}, address: {}]",
                                 clusterAlias,

diff --git a/server/src/main/java/org/opensearch/transport/RemoteClusterService.java b/server/src/main/java/org/opensearch/transport/RemoteClusterService.java
@@ -270,7 +270,7 @@ protected void updateRemoteCluster(String clusterAlias, Settings settings) {
             // 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));
+                logger.error("failed to connect to new remote cluster {} within {}", clusterAlias, TimeValue.timeValueSeconds(10));
             }
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();