Consolidate settings consumers for RemoteClusterService

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

@@ -9,8 +9,6 @@
 
 package org.elasticsearch.transport;
 
-import org.elasticsearch.cluster.metadata.ProjectId;
-
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 
@@ -29,15 +27,4 @@ public void register(LinkedProjectConfigListener listener) {
     protected void handleUpdate(LinkedProjectConfig config) {
         listeners.forEach(listener -> listener.updateLinkedProject(config));
     }
-
-    protected void handleSkipUnavailableChanged(
-        ProjectId originProjectId,
-        ProjectId linkedProjectId,
-        String linkedProjectAlias,
-        boolean skipUnavailable
-    ) {
-        listeners.forEach(
-            listener -> listener.skipUnavailableChanged(originProjectId, linkedProjectId, linkedProjectAlias, skipUnavailable)
-        );
-    }
 }

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

@@ -48,6 +48,7 @@ public ClusterSettingsLinkedProjectConfigService(
                 RemoteClusterSettings.REMOTE_CLUSTER_COMPRESS,
                 RemoteClusterSettings.REMOTE_CLUSTER_PING_SCHEDULE,
                 RemoteClusterSettings.REMOTE_CONNECTION_MODE,
+                RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
                 RemoteClusterSettings.SniffConnectionStrategySettings.REMOTE_CLUSTERS_PROXY,
                 RemoteClusterSettings.SniffConnectionStrategySettings.REMOTE_CLUSTER_SEEDS,
                 RemoteClusterSettings.SniffConnectionStrategySettings.REMOTE_NODE_CONNECTIONS,
@@ -56,11 +57,6 @@ public ClusterSettingsLinkedProjectConfigService(
                 RemoteClusterSettings.ProxyConnectionStrategySettings.SERVER_NAME
             );
             clusterSettings.addAffixGroupUpdateConsumer(remoteClusterSettings, this::settingsChangedCallback);
-            clusterSettings.addAffixUpdateConsumer(
-                RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
-                this::skipUnavailableChangedCallback,
-                (alias, value) -> {}
-            );
         }
     }
 
@@ -79,9 +75,4 @@ private void settingsChangedCallback(String clusterAlias, Settings newSettings)
         final var config = RemoteClusterSettings.toConfig(projectResolver.getProjectId(), ProjectId.DEFAULT, clusterAlias, mergedSettings);
         handleUpdate(config);
     }
-
-    @FixForMultiProject(description = "Refactor to add the linked project ID associated with the alias.")
-    private void skipUnavailableChangedCallback(String clusterAlias, Boolean skipUnavailable) {
-        handleSkipUnavailableChanged(projectResolver.getProjectId(), ProjectId.DEFAULT, clusterAlias, skipUnavailable);
-    }
 }

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

@@ -9,8 +9,6 @@
 
 package org.elasticsearch.transport;
 
-import org.elasticsearch.cluster.metadata.ProjectId;
-
 import java.util.Collection;
 
 /**
@@ -29,22 +27,6 @@ interface LinkedProjectConfigListener {
          * @param config The updated {@link LinkedProjectConfig}.
          */
         void updateLinkedProject(LinkedProjectConfig config);
-
-        /**
-         * Called when the boolean skip_unavailable setting has changed for a linked project configuration.
-         * Note that skip_unavailable may not be supported in all contexts where linked projects are used.
-         *
-         * @param originProjectId The {@link ProjectId} of the owning project that has the linked project configuration.
-         * @param linkedProjectId The {@link ProjectId} of the linked project.
-         * @param linkedProjectAlias The alias used for the linked project.
-         * @param skipUnavailable The new value of the skip_unavailable setting.
-         */
-        default void skipUnavailableChanged(
-            ProjectId originProjectId,
-            ProjectId linkedProjectId,
-            String linkedProjectAlias,
-            boolean skipUnavailable
-        ) {}
     }
 
     /**

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

@@ -285,19 +285,6 @@ public RemoteClusterConnection getRemoteClusterConnection(String cluster) {
         return connection;
     }
 
-    @Override
-    public void skipUnavailableChanged(
-        ProjectId originProjectId,
-        ProjectId linkedProjectId,
-        String linkedProjectAlias,
-        boolean skipUnavailable
-    ) {
-        final var remote = getConnectionsMapForProject(originProjectId).get(linkedProjectAlias);
-        if (remote != null) {
-            remote.setSkipUnavailable(skipUnavailable);
-        }
-    }
-
     @FixForMultiProject(description = "Refactor as needed to support project specific changes to linked remotes.")
     public synchronized void updateRemoteClusterCredentials(Supplier<Settings> settingsSupplier, ActionListener<Void> listener) {
         final var projectId = projectResolver.getProjectId();
@@ -450,6 +437,9 @@ synchronized void updateRemoteCluster(
             connectionMap.put(clusterAlias, remote);
             remote.ensureConnected(listener.map(ignored -> RemoteClusterConnectionStatus.RECONNECTED));
         } else {
+            if (crossProjectEnabled == false) {
+                remote.setSkipUnavailable(config.skipUnavailable());
+            }
             // No changes to connection configuration.
             listener.onResponse(RemoteClusterConnectionStatus.UNCHANGED);
         }

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

@@ -9,7 +9,6 @@
 
 package org.elasticsearch.transport;
 
-import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.project.DefaultProjectResolver;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
@@ -25,35 +24,16 @@
 
 public class ClusterSettingsLinkedProjectConfigServiceTests extends ESTestCase {
 
-    private record SkipUnavailableUpdate(
-        ProjectId originProjectId,
-        ProjectId linkedProjectId,
-        String linkedProjectAlias,
-        boolean skipUnavailable
-    ) {}
-
     private static class StubLinkedProjectConfigListener implements LinkedProjectConfigListener {
         LinkedProjectConfig updatedConfig;
-        SkipUnavailableUpdate skipUnavailableUpdate;
 
         @Override
         public void updateLinkedProject(LinkedProjectConfig config) {
             updatedConfig = config;
         }
 
-        @Override
-        public void skipUnavailableChanged(
-            ProjectId originProjectId,
-            ProjectId linkedProjectId,
-            String linkedProjectAlias,
-            boolean skipUnavailable
-        ) {
-            skipUnavailableUpdate = new SkipUnavailableUpdate(originProjectId, linkedProjectId, linkedProjectAlias, skipUnavailable);
-        }
-
         void reset() {
             updatedConfig = null;
-            skipUnavailableUpdate = null;
         }
     }
 
@@ -66,17 +46,21 @@ public void testListenersReceiveUpdates() {
         final var alias = randomAlphaOfLength(10);
 
         final var initialProxyAddress = "localhost:9400";
+        final var initialSkipUnavailable = randomBoolean();
         final var initialSettings = Settings.builder()
             .put("cluster.remote." + alias + ".mode", "proxy")
             .put("cluster.remote." + alias + ".proxy_address", initialProxyAddress)
+            .put("cluster.remote." + alias + ".skip_unavailable", initialSkipUnavailable)
             .build();
         final var clusterSettings = ClusterSettings.createBuiltInClusterSettings(initialSettings);
         final var service = new ClusterSettingsLinkedProjectConfigService(
             initialSettings,
             clusterSettings,
             DefaultProjectResolver.INSTANCE
         );
-        final var config = new ProxyLinkedProjectConfigBuilder(alias).proxyAddress(initialProxyAddress).build();
+        final var config = new ProxyLinkedProjectConfigBuilder(alias).proxyAddress(initialProxyAddress)
+            .skipUnavailable(initialSkipUnavailable)
+            .build();
 
         // Verify we can get the linked projects on startup.
         assertThat(service.getInitialLinkedProjectConfigs(), equalTo(List.of(config)));
@@ -92,48 +76,23 @@ public void testListenersReceiveUpdates() {
         clusterSettings.applySettings(initialSettings);
         for (int i = 0; i < numListeners; ++i) {
             assertThat(listeners.get(i).updatedConfig, sameInstance(null));
-            assertThat(listeners.get(i).skipUnavailableUpdate, sameInstance(null));
             listeners.get(i).reset();
         }
 
-        // Change the skip_unavailable, leave the other settings alone, we should get the skip_unavailable update only.
-        var expectedSkipUnavailableUpdate = new SkipUnavailableUpdate(
-            config.originProjectId(),
-            config.linkedProjectId(),
-            config.linkedProjectAlias(),
-            config.skipUnavailable() == false
-        );
-        clusterSettings.applySettings(
-            Settings.builder()
-                .put(initialSettings)
-                .put("cluster.remote." + alias + ".skip_unavailable", expectedSkipUnavailableUpdate.skipUnavailable)
-                .build()
-        );
-        for (int i = 0; i < numListeners; ++i) {
-            assertThat(listeners.get(i).updatedConfig, sameInstance(null));
-            assertThat(listeners.get(i).skipUnavailableUpdate, equalTo(expectedSkipUnavailableUpdate));
-            listeners.get(i).reset();
-        }
-
-        // Change the proxy address, and set skip_unavailable back to original value, we should get both updates.
-        expectedSkipUnavailableUpdate = new SkipUnavailableUpdate(
-            config.originProjectId(),
-            config.linkedProjectId(),
-            config.linkedProjectAlias(),
-            config.skipUnavailable()
-        );
+        // Change the proxy address and skip_unavailable values.
         final var newProxyAddress = "localhost:9401";
+        final var newSkipUnavailable = config.skipUnavailable() == false;
         clusterSettings.applySettings(
             Settings.builder()
                 .put(initialSettings)
                 .put("cluster.remote." + alias + ".proxy_address", newProxyAddress)
-                .put("cluster.remote." + alias + ".skip_unavailable", expectedSkipUnavailableUpdate.skipUnavailable)
+                .put("cluster.remote." + alias + ".skip_unavailable", newSkipUnavailable)
                 .build()
         );
         for (int i = 0; i < numListeners; ++i) {
             assertNotNull("expected non-null updatedConfig for listener " + i, listeners.get(i).updatedConfig);
             assertThat(listeners.get(i).updatedConfig.proxyAddress(), equalTo(newProxyAddress));
-            assertThat(listeners.get(i).skipUnavailableUpdate, equalTo(expectedSkipUnavailableUpdate));
+            assertThat(listeners.get(i).updatedConfig.skipUnavailable(), equalTo(newSkipUnavailable));
         }
     }
 }

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

@@ -51,6 +51,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 
 import static org.elasticsearch.test.MockLog.assertThatLogger;
@@ -111,6 +112,10 @@ private MockTransportService startTransport(
         return RemoteClusterConnectionTests.startTransport(id, knownNodes, version, transportVersion, threadPool, settings);
     }
 
+    private MockTransportService startTransport(final String id) {
+        return startTransport(id, List.of(), VersionInformation.CURRENT, TransportVersion.current(), Settings.EMPTY);
+    }
+
     public void testSettingsAreRegistered() {
         assertTrue(ClusterSettings.BUILT_IN_CLUSTER_SETTINGS.contains(RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE));
         assertTrue(ClusterSettings.BUILT_IN_CLUSTER_SETTINGS.contains(RemoteClusterSettings.REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING));
@@ -1828,6 +1833,48 @@ public void testLogsConnectionResult() throws IOException {
         }
     }
 
+    public void testSetSkipUnavailable() throws IOException {
+        final var skipUnavailableProperty = RemoteClusterSettings.REMOTE_CLUSTER_SKIP_UNAVAILABLE.getConcreteSettingForNamespace("remote")
+            .getKey();
+        final var seedNodeProperty = SniffConnectionStrategySettings.REMOTE_CLUSTER_SEEDS.getConcreteSettingForNamespace("remote").getKey();
+        final var clusterSettings = ClusterSettings.createBuiltInClusterSettings();
+
+        try (
+            var remote1Transport = startTransport("remote1");
+            var remote2Transport = startTransport("remote2");
+            var local = startTransport("local");
+            var remoteClusterService = createRemoteClusterService(Settings.EMPTY, clusterSettings, local)
+        ) {
+            linkedProjectConfigService.register(remoteClusterService);
+
+            final BiConsumer<Boolean, MockTransportService> applySettingsAndVerify = (skipUnavailableValue, mockTransportService) -> {
+                clusterSettings.applySettings(
+                    Settings.builder()
+                        .put(skipUnavailableProperty, skipUnavailableValue)
+                        .put(seedNodeProperty, mockTransportService.getLocalNode().getAddress().toString())
+                        .build()
+                );
+                assertTrue(isRemoteClusterRegistered(remoteClusterService, "remote"));
+                assertEquals(skipUnavailableValue, remoteClusterService.isSkipUnavailable("remote").orElseThrow());
+            };
+
+            // Apply the initial settings and verify the new connection is built.
+            var skipUnavailable = randomBoolean();
+            applySettingsAndVerify.accept(skipUnavailable, remote1Transport);
+
+            // Change skip_unavailable value, but not seed node, connection should not be rebuilt, but skip_unavailable should be modified.
+            skipUnavailable = skipUnavailable == false;
+            applySettingsAndVerify.accept(skipUnavailable, remote1Transport);
+
+            // Change the seed node but not skip_unavailable, connection should be rebuilt and skip_unavailable should stay the same.
+            applySettingsAndVerify.accept(skipUnavailable, remote2Transport);
+
+            // Change skip_unavailable value and the seed node, connection should be rebuilt and skip_unavailable should also be modified.
+            skipUnavailable = skipUnavailable == false;
+            applySettingsAndVerify.accept(skipUnavailable, remote1Transport);
+        }
+    }
+
     private void updateRemoteCluster(
         RemoteClusterService service,
         String alias,