Update RemoteClusterService skip_unavailable handling for CPS (#132478)

In serverless cross-project search (CPS) skip_unavailable will not be used and should always be
true. This PR adds a settings validator that rejects setting skip_unavailable when CPS is enabled.

Resolves: ES-12267

Author: JeremyDahlgren

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

@@ -300,6 +300,8 @@ public void skipUnavailableChanged(
         String linkedProjectAlias,
         boolean skipUnavailable
     ) {
+        assert crossProjectEnabled == false
+            : "Cannot configure setting [" + RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE.getKey() + "] in CPS environments.";
         final var remote = getConnectionsMapForProject(originProjectId).get(linkedProjectAlias);
         if (remote != null) {
             remote.setSkipUnavailable(skipUnavailable);

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

@@ -13,6 +13,7 @@
 import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.settings.SecureSetting;
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Setting;
@@ -70,7 +71,7 @@ public class RemoteClusterSettings {
         (ns, key) -> boolSetting(
             key,
             DEFAULT_SKIP_UNAVAILABLE,
-            new RemoteConnectionEnabled<>(ns, key),
+            new UnsupportedInCPSValidator<>(ns, key),
             Setting.Property.Dynamic,
             Setting.Property.NodeScope
         )
@@ -390,6 +391,10 @@ private RemoteConnectionEnabled(String clusterAlias, String key) {
             this.key = key;
         }
 
+        protected String getKey() {
+            return key;
+        }
+
         @Override
         public void validate(T value) {}
 
@@ -483,4 +488,25 @@ public Iterator<Setting<?>> settings() {
             return settingStream.iterator();
         }
     }
+
+    private static class UnsupportedInCPSValidator<T> extends RemoteConnectionEnabled<T> {
+        private final Setting<Boolean> cpsSetting = Setting.boolSetting("serverless.cross_project.enabled", false);
+
+        private UnsupportedInCPSValidator(String clusterAlias, String key) {
+            super(clusterAlias, key);
+        }
+
+        @Override
+        public void validate(T value, Map<Setting<?>, Object> settings, boolean isPresent) {
+            if (isPresent && (Boolean) settings.get(cpsSetting)) {
+                throw new IllegalArgumentException("setting [" + getKey() + "] is unavailable when CPS is enabled");
+            }
+            super.validate(value, settings, isPresent);
+        }
+
+        @Override
+        public Iterator<Setting<?>> settings() {
+            return Iterators.concat(super.settings(), Iterators.single(cpsSetting));
+        }
+    }
 }

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

@@ -22,8 +22,10 @@
 
 import static org.elasticsearch.test.NodeRoles.nonRemoteClusterClientNode;
 import static org.elasticsearch.test.NodeRoles.remoteClusterClientNode;
+import static org.elasticsearch.transport.RemoteClusterSettings.ProxyConnectionStrategySettings.PROXY_ADDRESS;
 import static org.elasticsearch.transport.RemoteClusterSettings.REMOTE_CLUSTER_CREDENTIALS;
 import static org.elasticsearch.transport.RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE;
+import static org.elasticsearch.transport.RemoteClusterSettings.REMOTE_CONNECTION_MODE;
 import static org.elasticsearch.transport.RemoteClusterSettings.REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING;
 import static org.elasticsearch.transport.RemoteClusterSettings.REMOTE_NODE_ATTRIBUTE;
 import static org.elasticsearch.transport.RemoteClusterSettings.SniffConnectionStrategySettings.REMOTE_CLUSTERS_PROXY;
@@ -95,4 +97,51 @@ public void testProxyDefault() {
         final String alias = randomAlphaOfLength(8);
         assertThat(REMOTE_CLUSTERS_PROXY.getConcreteSettingForNamespace(alias).get(Settings.EMPTY), equalTo(""));
     }
+
+    public void testSkipUnavailableAlwaysTrueIfCPSEnabled() {
+        final var alias = randomAlphaOfLength(8);
+        final var skipUnavailableSetting = REMOTE_CLUSTER_SKIP_UNAVAILABLE.getConcreteSettingForNamespace(alias);
+        final var modeSetting = REMOTE_CONNECTION_MODE.getConcreteSettingForNamespace(alias);
+        final var proxyAddressSetting = PROXY_ADDRESS.getConcreteSettingForNamespace(alias);
+        final var cpsEnabledSettings = Settings.builder().put("serverless.cross_project.enabled", 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-CPS 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-CPS 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 CPS is enabled.
+        exception = expectThrows(
+            IllegalArgumentException.class,
+            () -> skipUnavailableSetting.get(
+                Settings.builder()
+                    .put(cpsEnabledSettings)
+                    .put(proxyEnabledSettings)
+                    .put(skipUnavailableSetting.getKey(), randomBoolean())
+                    .build()
+            )
+        );
+        assertThat(exception.getMessage(), equalTo("setting [" + skipUnavailableSetting.getKey() + "] is unavailable when CPS is enabled"));
+
+        // Should not throw if the setting is not present, returning the expected default value of true.
+        assertTrue(skipUnavailableSetting.get(Settings.builder().put(cpsEnabledSettings).put(proxyEnabledSettings).build()));
+    }
 }