Update RemoteClusterService to default skip unavailable to true for CPS

In stateless CPS skip_unavailable will not be used and should always be
true.  Also DisconnectedStrategy.RECONNECT_UNLESS_SKIP_UNAVAILABLE is
rejected in CPS as a parameter value for getRemoteClusterClient().

Resolves: ES-12267

Author: JeremyDahlgren

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

@@ -52,6 +52,7 @@ public final class RemoteClusterConnection implements Closeable {
     private final ThreadPool threadPool;
     private volatile boolean skipUnavailable;
     private final TimeValue initialConnectionTimeout;
+    private final boolean isStateless;
 
     /**
      * Creates a new {@link RemoteClusterConnection}
@@ -83,8 +84,9 @@ public final class RemoteClusterConnection implements Closeable {
         this.connectionStrategy = RemoteConnectionStrategy.buildStrategy(clusterAlias, transportService, remoteConnectionManager, settings);
         // we register the transport service here as a listener to make sure we notify handlers on disconnect etc.
         this.remoteConnectionManager.addListener(transportService);
-        this.skipUnavailable = RemoteClusterService.REMOTE_CLUSTER_SKIP_UNAVAILABLE.getConcreteSettingForNamespace(clusterAlias)
-            .get(settings);
+        this.isStateless = DiscoveryNode.isStateless(settings);
+        this.skipUnavailable = isStateless
+            || RemoteClusterService.REMOTE_CLUSTER_SKIP_UNAVAILABLE.getConcreteSettingForNamespace(clusterAlias).get(settings);
         this.threadPool = transportService.threadPool;
         initialConnectionTimeout = RemoteClusterService.REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING.get(settings);
     }
@@ -93,6 +95,7 @@ public final class RemoteClusterConnection implements Closeable {
      * Updates the skipUnavailable flag that can be dynamically set for each remote cluster
      */
     void setSkipUnavailable(boolean skipUnavailable) {
+        assert isStateless == false : "Cannot set skipUnavailable in stateless environments.";
         this.skipUnavailable = skipUnavailable;
     }
 

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

@@ -149,6 +149,7 @@ public final class RemoteClusterService extends RemoteClusterAware
 
     private final boolean enabled;
     private final boolean remoteClusterServerEnabled;
+    private final boolean isStateless;
 
     public boolean isEnabled() {
         return enabled;
@@ -167,6 +168,7 @@ public boolean isRemoteClusterServerEnabled() {
     RemoteClusterService(Settings settings, TransportService transportService) {
         super(settings);
         this.enabled = DiscoveryNode.isRemoteClusterClient(settings);
+        this.isStateless = DiscoveryNode.isStateless(settings);
         this.remoteClusterServerEnabled = REMOTE_CLUSTER_SERVER_ENABLED.get(settings);
         this.transportService = transportService;
         this.projectResolver = DefaultProjectResolver.INSTANCE;
@@ -294,7 +296,7 @@ void ensureConnected(String clusterAlias, ActionListener<Void> listener) {
      * Returns whether the cluster identified by the provided alias is configured to be skipped when unavailable
      */
     public boolean isSkipUnavailable(String clusterAlias) {
-        return getRemoteClusterConnection(clusterAlias).isSkipUnavailable();
+        return isStateless || getRemoteClusterConnection(clusterAlias).isSkipUnavailable();
     }
 
     public Transport.Connection getConnection(String cluster) {
@@ -351,10 +353,13 @@ public RemoteClusterConnection getRemoteClusterConnection(String cluster) {
     @Override
     public void listenForUpdates(ClusterSettings clusterSettings) {
         super.listenForUpdates(clusterSettings);
-        clusterSettings.addAffixUpdateConsumer(REMOTE_CLUSTER_SKIP_UNAVAILABLE, this::updateSkipUnavailable, (alias, value) -> {});
+        if (isStateless == false) {
+            clusterSettings.addAffixUpdateConsumer(REMOTE_CLUSTER_SKIP_UNAVAILABLE, this::updateSkipUnavailable, (alias, value) -> {});
+        }
     }
 
     private synchronized void updateSkipUnavailable(String clusterAlias, Boolean skipUnavailable) {
+        assert isStateless == false : "Cannot configure setting [" + REMOTE_CLUSTER_SKIP_UNAVAILABLE + "] in stateless environments.";
         RemoteClusterConnection remote = getConnectionsMapForCurrentProject().get(clusterAlias);
         if (remote != null) {
             remote.setSkipUnavailable(skipUnavailable);
@@ -667,6 +672,13 @@ public RemoteClusterClient getRemoteClusterClient(
                 "this node does not have the " + DiscoveryNodeRole.REMOTE_CLUSTER_CLIENT_ROLE.roleName() + " role"
             );
         }
+        if (isStateless && disconnectedStrategy == DisconnectedStrategy.RECONNECT_UNLESS_SKIP_UNAVAILABLE) {
+            throw new IllegalArgumentException(
+                "DisconnectedStrategy ["
+                    + DisconnectedStrategy.RECONNECT_UNLESS_SKIP_UNAVAILABLE
+                    + "] is not supported in stateless environments"
+            );
+        }
         if (transportService.getRemoteClusterService().getRegisteredRemoteClusterNames().contains(clusterAlias) == false) {
             throw new NoSuchRemoteClusterException(clusterAlias);
         }

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

@@ -240,6 +240,29 @@ public void testRemoteClusterServiceNotEnabled() {
         }
     }
 
+    public void testGetRemoteClusterClientRejectsSkipUnavailableInStateless() {
+        final var settings = Settings.builder().put("stateless.enabled", true).putList("node.roles", "remote_cluster_client").build();
+        try (
+            MockTransportService service = MockTransportService.createNewService(
+                settings,
+                VersionInformation.CURRENT,
+                TransportVersion.current(),
+                threadPool,
+                null
+            )
+        ) {
+            final var remoteClusterService = service.getRemoteClusterService();
+            expectThrows(
+                IllegalArgumentException.class,
+                () -> remoteClusterService.getRemoteClusterClient(
+                    "test",
+                    EsExecutors.DIRECT_EXECUTOR_SERVICE,
+                    RemoteClusterService.DisconnectedStrategy.RECONNECT_UNLESS_SKIP_UNAVAILABLE
+                )
+            );
+        }
+    }
+
     public void testQuicklySkipUnavailableClusters() throws Exception {
         Settings remoteSettings = Settings.builder().put(ClusterName.CLUSTER_NAME_SETTING.getKey(), "foo_bar_cluster").build();
         try (