Give helpful message on remote connections disabled (#53690)

Today when cluster.remote.connect is set to false, and some aspect of
the codebase tries to get a remote client, today we return a no such
remote cluster exception. This can be quite perplexing to users,
especially if the remote cluster is actually defined in their cluster
state, it is only that the local node is not a remote cluter
client. This commit addresses this by providing a dedicated error
message when a remote cluster is not available because the local node is
not a remote cluster client.

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

@@ -164,11 +164,18 @@ public String getKey(final String key) {
         (ns, key) -> boolSetting(key, TransportSettings.TRANSPORT_COMPRESS,
             new RemoteConnectionEnabled<>(ns, key), Setting.Property.Dynamic, Setting.Property.NodeScope));
 
+    private final boolean enabled;
+
+    public boolean isEnabled() {
+        return enabled;
+    }
+
     private final TransportService transportService;
     private final Map<String, RemoteClusterConnection> remoteClusters = ConcurrentCollections.newConcurrentMap();
 
     RemoteClusterService(Settings settings, TransportService transportService) {
         super(settings);
+        this.enabled = ENABLE_REMOTE_CLUSTERS.get(settings);
         this.transportService = transportService;
     }
 
@@ -248,6 +255,9 @@ public Transport.Connection getConnection(String cluster) {
     }
 
     RemoteClusterConnection getRemoteClusterConnection(String cluster) {
+        if (enabled == false) {
+            throw new IllegalArgumentException("remote cluster service is not enabled");
+        }
         RemoteClusterConnection connection = remoteClusters.get(cluster);
         if (connection == null) {
             throw new NoSuchRemoteClusterException(cluster);
@@ -385,6 +395,9 @@ public Stream<RemoteConnectionInfo> getRemoteConnectionInfos() {
      * function on success.
      */
     public void collectNodes(Set<String> clusters, ActionListener<BiFunction<String, String, DiscoveryNode>> listener) {
+        if (enabled == false) {
+            throw new IllegalArgumentException("remote cluster service is not enabled");
+        }
         Map<String, RemoteClusterConnection> remoteClusters = this.remoteClusters;
         for (String cluster : clusters) {
             if (remoteClusters.containsKey(cluster) == false) {
@@ -428,6 +441,9 @@ public void onFailure(Exception e) {
      * @throws IllegalArgumentException if the given clusterAlias doesn't exist
      */
     public Client getRemoteClusterClient(ThreadPool threadPool, String clusterAlias) {
+        if (transportService.getRemoteClusterService().isEnabled() == false) {
+            throw new IllegalArgumentException("remote cluster service is not enabled");
+        }
         if (transportService.getRemoteClusterService().getRemoteClusterNames().contains(clusterAlias) == false) {
             throw new NoSuchRemoteClusterException(clusterAlias);
         }

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

@@ -34,6 +34,7 @@
 import java.util.concurrent.TimeUnit;
 
 import static org.elasticsearch.transport.RemoteClusterConnectionTests.startTransport;
+import static org.hamcrest.Matchers.equalTo;
 
 public class RemoteClusterClientTests extends ESTestCase {
     private final ThreadPool threadPool = new TestThreadPool(getClass().getName());
@@ -119,4 +120,17 @@ public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connecti
             }
         }
     }
+
+    public void testRemoteClusterServiceNotEnabled() {
+        final Settings settings = Settings.builder().put(RemoteClusterService.ENABLE_REMOTE_CLUSTERS.getKey(), false).build();
+        try (MockTransportService service = MockTransportService.createNewService(settings, Version.CURRENT, threadPool, null)) {
+            service.start();
+            service.acceptIncomingRequests();
+            final RemoteClusterService remoteClusterService = service.getRemoteClusterService();
+            final IllegalArgumentException e =
+                expectThrows(IllegalArgumentException.class, () -> remoteClusterService.getRemoteClusterClient(threadPool, "test"));
+            assertThat(e.getMessage(), equalTo("remote cluster service is not enabled"));
+        }
+    }
+
 }

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

@@ -42,6 +42,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -850,10 +851,34 @@ public void testSkipUnavailable() {
         }
     }
 
+    public void testRemoteClusterServiceNotEnabledGetRemoteClusterConnection() {
+        final Settings settings = Settings.builder().put(RemoteClusterService.ENABLE_REMOTE_CLUSTERS.getKey(), false).build();
+        try (MockTransportService service = MockTransportService.createNewService(settings, Version.CURRENT, threadPool, null)) {
+            service.start();
+            service.acceptIncomingRequests();
+            final IllegalArgumentException e =
+                expectThrows(IllegalArgumentException.class, () -> service.getRemoteClusterService().getRemoteClusterConnection("test"));
+            assertThat(e.getMessage(), equalTo("remote cluster service is not enabled"));
+        }
+    }
+
+    public void testRemoteClusterServiceNotEnabledGetCollectNodes() {
+        final Settings settings = Settings.builder().put(RemoteClusterService.ENABLE_REMOTE_CLUSTERS.getKey(), false).build();
+        try (MockTransportService service = MockTransportService.createNewService(settings, Version.CURRENT, threadPool, null)) {
+            service.start();
+            service.acceptIncomingRequests();
+            final IllegalArgumentException e = expectThrows(
+                IllegalArgumentException.class,
+                () -> service.getRemoteClusterService().collectNodes(Set.of(), ActionListener.wrap(r -> {}, r -> {})));
+            assertThat(e.getMessage(), equalTo("remote cluster service is not enabled"));
+        }
+    }
+
     private static Settings createSettings(String clusterAlias, List<String> seeds) {
         Settings.Builder builder = Settings.builder();
         builder.put(SniffConnectionStrategy.REMOTE_CLUSTER_SEEDS.getConcreteSettingForNamespace(clusterAlias).getKey(),
             Strings.collectionToCommaDelimitedString(seeds));
         return builder.build();
     }
+
 }