Update RemoteClusterService skip_unavailable handling for CPS

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

@@ -27,6 +27,7 @@
 import org.elasticsearch.cluster.project.DefaultProjectResolver;
 import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.SecureSetting;
 import org.elasticsearch.common.settings.SecureString;
@@ -98,7 +99,13 @@ public final class RemoteClusterService extends RemoteClusterAware
     public static final Setting.AffixSetting<Boolean> REMOTE_CLUSTER_SKIP_UNAVAILABLE = Setting.affixKeySetting(
         "cluster.remote.",
         "skip_unavailable",
-        (ns, key) -> boolSetting(key, true, new RemoteConnectionEnabled<>(ns, key), Setting.Property.Dynamic, Setting.Property.NodeScope)
+        (ns, key) -> boolSetting(
+            key,
+            true,
+            new UnsupportedInStatelessValidator<>(ns, key),
+            Setting.Property.Dynamic,
+            Setting.Property.NodeScope
+        )
     );
 
     public static final Setting.AffixSetting<TimeValue> REMOTE_CLUSTER_PING_SCHEDULE = Setting.affixKeySetting(
@@ -343,10 +350,14 @@ 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.getKey() + "] in stateless environments.";
         RemoteClusterConnection remote = getConnectionsMapForCurrentProject().get(clusterAlias);
         if (remote != null) {
             remote.setSkipUnavailable(skipUnavailable);
@@ -748,6 +759,10 @@ private RemoteConnectionEnabled(String clusterAlias, String key) {
             this.key = key;
         }
 
+        protected String getKey() {
+            return key;
+        }
+
         @Override
         public void validate(T value) {}
 
@@ -772,4 +787,25 @@ private Stream<Setting<?>> settingsStream() {
                 .map(as -> as.getConcreteSettingForNamespace(clusterAlias));
         }
     };
+
+    private static class UnsupportedInStatelessValidator<T> extends RemoteConnectionEnabled<T> {
+        private final Setting<Boolean> statelessSetting = Setting.boolSetting(DiscoveryNode.STATELESS_ENABLED_SETTING_NAME, false);
+
+        private UnsupportedInStatelessValidator(String clusterAlias, String key) {
+            super(clusterAlias, key);
+        }
+
+        @Override
+        public void validate(T value, Map<Setting<?>, Object> settings, boolean isPresent) {
+            if (isPresent && (Boolean) settings.get(statelessSetting)) {
+                throw new IllegalArgumentException("setting [" + getKey() + "] is unavailable when stateless is enabled");
+            }
+            super.validate(value, settings, isPresent);
+        }
+
+        @Override
+        public Iterator<Setting<?>> settings() {
+            return Iterators.concat(super.settings(), Iterators.single(statelessSetting));
+        }
+    }
 }

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

@@ -95,4 +95,54 @@ public void testProxyDefault() {
         final String alias = randomAlphaOfLength(8);
         assertThat(REMOTE_CLUSTERS_PROXY.getConcreteSettingForNamespace(alias).get(Settings.EMPTY), equalTo(""));
     }
+
+    public void testSkipUnavailableAlwaysTrueIfStatelessEnabled() {
+        final var alias = randomAlphaOfLength(8);
+        final var skipUnavailableSetting = REMOTE_CLUSTER_SKIP_UNAVAILABLE.getConcreteSettingForNamespace(alias);
+        final var modeSetting = RemoteConnectionStrategy.REMOTE_CONNECTION_MODE.getConcreteSettingForNamespace(alias);
+        final var proxyAddressSetting = ProxyConnectionStrategy.PROXY_ADDRESS.getConcreteSettingForNamespace(alias);
+        final var statelessEnabledSettings = Settings.builder().put(DiscoveryNode.STATELESS_ENABLED_SETTING_NAME, true).build();
+        final var proxyEnabledSettings = Settings.builder()
+            .put(modeSetting.getKey(), RemoteConnectionStrategy.ConnectionStrategy.PROXY.toString())
+            .put(proxyAddressSetting.getKey(), "localhost:9400")
+            .build();
+
+        // Ensure the validator still throws in non-stateless environment if a connection mode is not set.
+        var exception = expectThrows(
+            IllegalArgumentException.class,
+            () -> skipUnavailableSetting.get(Settings.builder().put(skipUnavailableSetting.getKey(), true).build())
+        );
+        assertThat(
+            exception.getMessage(),
+            equalTo("Cannot configure setting [" + skipUnavailableSetting.getKey() + "] if remote cluster is not enabled.")
+        );
+
+        // Ensure we can still get the set value in non-stateless environment.
+        final var randomSkipUnavailableSettingValue = randomBoolean();
+        assertThat(
+            skipUnavailableSetting.get(
+                Settings.builder().put(proxyEnabledSettings).put(skipUnavailableSetting.getKey(), randomSkipUnavailableSettingValue).build()
+            ),
+            equalTo(randomSkipUnavailableSettingValue)
+        );
+
+        // Check the validator rejects the skip_unavailable setting if present when stateless is enabled.
+        exception = expectThrows(
+            IllegalArgumentException.class,
+            () -> skipUnavailableSetting.get(
+                Settings.builder()
+                    .put(statelessEnabledSettings)
+                    .put(proxyEnabledSettings)
+                    .put(skipUnavailableSetting.getKey(), randomBoolean())
+                    .build()
+            )
+        );
+        assertThat(
+            exception.getMessage(),
+            equalTo("setting [" + skipUnavailableSetting.getKey() + "] is unavailable when stateless is enabled")
+        );
+
+        // Should not throw if the setting is not present, returning the expected default value of true.
+        assertTrue(skipUnavailableSetting.get(Settings.builder().put(statelessEnabledSettings).put(proxyEnabledSettings).build()));
+    }
 }