/*
 * Licensed to Elasticsearch under one or more contributor
 * license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright
 * ownership. Elasticsearch licenses this file to you under
 * the Apache License, Version 2.0 (the "License"); you may
 * not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.elasticsearch.transport;

import org.elasticsearch.Version;
import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterName;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.test.ESTestCase;
import org.elasticsearch.test.transport.MockTransportService;
import org.elasticsearch.threadpool.TestThreadPool;
import org.elasticsearch.threadpool.ThreadPool;

import java.util.Collections;
import java.util.concurrent.TimeUnit;

import static org.elasticsearch.transport.RemoteClusterConnectionTests.startTransport;

public class RemoteClusterClientTests extends ESTestCase {
    private final ThreadPool threadPool = new TestThreadPool(getClass().getName());

    @Override
    public void tearDown() throws Exception {
        super.tearDown();
        ThreadPool.terminate(threadPool, 10, TimeUnit.SECONDS);
    }

    public void testConnectAndExecuteRequest() throws Exception {
        Settings remoteSettings = Settings.builder().put(ClusterName.CLUSTER_NAME_SETTING.getKey(), "foo_bar_cluster").build();
        try (MockTransportService remoteTransport = startTransport("remote_node", Collections.emptyList(), Version.CURRENT, threadPool,
            remoteSettings)) {
            DiscoveryNode remoteNode = remoteTransport.getLocalDiscoNode();

            Settings localSettings = Settings.builder()
                .put(RemoteClusterService.ENABLE_REMOTE_CLUSTERS.getKey(), true)
                .put("search.remote.test.seeds", remoteNode.getAddress().getAddress() + ":" + remoteNode.getAddress().getPort()).build();
            try (MockTransportService service = MockTransportService.createNewService(localSettings, Version.CURRENT, threadPool, null)) {
                service.start();
                service.acceptIncomingRequests();
                RemoteClusterService remoteClusterService = service.getRemoteClusterService();
                assertTrue(remoteClusterService.isRemoteNodeConnected("test", remoteNode));
                Client client = remoteClusterService.getRemoteClusterClient(threadPool, "test");
                ClusterStateResponse clusterStateResponse = client.admin().cluster().prepareState().execute().get();
                assertNotNull(clusterStateResponse);
                assertEquals("foo_bar_cluster", clusterStateResponse.getState().getClusterName().value());
                // also test a failure, there is no handler for search registered
                ActionNotFoundTransportException ex = expectThrows(ActionNotFoundTransportException.class,
                    () -> client.prepareSearch().get());
                assertEquals("No handler for action [indices:data/read/search]", ex.getMessage());
            }
        }
    }

    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/29547")
    public void testEnsureWeReconnect() throws Exception {
        Settings remoteSettings = Settings.builder().put(ClusterName.CLUSTER_NAME_SETTING.getKey(), "foo_bar_cluster").build();
        try (MockTransportService remoteTransport = startTransport("remote_node", Collections.emptyList(), Version.CURRENT, threadPool,
            remoteSettings)) {
            DiscoveryNode remoteNode = remoteTransport.getLocalDiscoNode();
            Settings localSettings = Settings.builder()
                .put(RemoteClusterService.ENABLE_REMOTE_CLUSTERS.getKey(), true)
                .put("search.remote.test.seeds", remoteNode.getAddress().getAddress() + ":" + remoteNode.getAddress().getPort()).build();
            try (MockTransportService service = MockTransportService.createNewService(localSettings, Version.CURRENT, threadPool, null)) {
                service.start();
                service.acceptIncomingRequests();
                service.disconnectFromNode(remoteNode);
                RemoteClusterService remoteClusterService = service.getRemoteClusterService();
                assertBusy(() -> assertFalse(remoteClusterService.isRemoteNodeConnected("test", remoteNode)));
                Client client = remoteClusterService.getRemoteClusterClient(threadPool, "test");
                ClusterStateResponse clusterStateResponse = client.admin().cluster().prepareState().execute().get();
                assertNotNull(clusterStateResponse);
                assertEquals("foo_bar_cluster", clusterStateResponse.getState().getClusterName().value());
            }
        }
    }

}

@@ -1353,16 +1353,55 @@
 ctions;%0A
+import java.util.concurrent.Semaphore;%0A
 import j
@@ -3696,89 +3696,8 @@
  %7D%0A%0A
-    @AwaitsFix(bugUrl = %22https://github.com/elastic/elasticsearch/issues/29547%22)%0A
     
@@ -4556,272 +4556,938 @@
     
-service.start();%0A                service.acceptIncomingRequests();%0A                service.disconnectFromNode(remoteNode);%0A                RemoteClusterService remoteClusterService = service.getRemoteClusterService();%0A                assertBusy(() -%3E assertFalse(r
+Semaphore semaphore = new Semaphore(1);%0A                service.start();%0A                service.addConnectionListener(new TransportConnectionListener() %7B%0A                    @Override%0A                    public void onNodeDisconnected(DiscoveryNode node) %7B%0A                        if (remoteNode.equals(node)) %7B%0A                            semaphore.release();%0A                        %7D%0A                    %7D%0A                %7D);%0A                // this test is not perfect since we might reconnect concurrently but it will fail most of the time if we don't have%0A                // the right calls in place in the RemoteAwareClient%0A                service.acceptIncomingRequests();%0A                for (int i = 0; i %3C 10; i++) %7B%0A                    semaphore.acquire();%0A                    try %7B%0A                        service.disconnectFromNode(remoteNode);%0A                        semaphore.acquire();%0A                        R
 emot
@@ -5505,54 +5505,75 @@
 vice
-.isR
+ r
 emote
-NodeConnected(%22test%22, remoteNode)));%0A
+ClusterService = service.getRemoteClusterService();%0A        
     
@@ -5673,32 +5673,40 @@
                 
+  
+      
 ClusterStateResp
@@ -5798,32 +5798,40 @@
                 
+        
 assertNotNull(cl
@@ -5859,32 +5859,40 @@
                 
+   
+     
 assertEquals(%22fo
@@ -5958,32 +5958,249 @@
 ame().value());%0A
+                        assertTrue(remoteClusterService.isRemoteNodeConnected(%22test%22, remoteNode));%0A                    %7D finally %7B%0A                        semaphore.release();%0A                    %7D%0A                %7D%0A
             %7D%0A  
@@ -6213,11 +6213,10 @@
 %7D%0A    %7D%0A
-%0A
 %7D%0A

Fix RemoteClusterClientTests#testEnsureWeReconnect