Use latest setting value when initializing setting watch (#134091)

When new cluster settings are applied, consumers watchings are notified.
The consumer should see the latest value at the time of initializing a
watch. However, only the node settings were ever used in finding the
value to initialize the consumer. This commit fixes the consumer to use
the value from latest applied if it exists.

closes #133701

docs/changelog/134091.yaml

@@ -0,0 +1,6 @@
+pr: 134091
+summary: Use latest setting value when initializing setting watch
+area: Infra/Settings
+type: bug
+issues:
+ - 133701

server/src/main/java/org/elasticsearch/common/settings/AbstractScopedSettings.java

@@ -433,7 +433,11 @@ public abstract class AbstractScopedSettings {
         assert setting.getProperties().contains(Setting.Property.Dynamic)
             || setting.getProperties().contains(Setting.Property.OperatorDynamic) : "Can only watch dynamic settings";
         assert setting.getProperties().contains(Setting.Property.NodeScope) : "Can only watch node settings";
-        consumer.accept(setting.get(settings));
+
+        // this mimics the combined settings of last applied and node settings, without building a new settings object
+        Settings settingsWithValue = setting.exists(lastSettingsApplied) ? lastSettingsApplied : settings;
+
+        consumer.accept(setting.get(settingsWithValue));
         addSettingsUpdateConsumer(setting, consumer);
     }
 

server/src/test/java/org/elasticsearch/action/admin/cluster/settings/ClusterSettingsTests.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.admin.cluster.settings;
+
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.Set;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class ClusterSettingsTests extends ESTestCase {
+
+    public void testWatchAfterApply() {
+        Setting<String> clusterSetting = Setting.simpleString("cluster.setting", Setting.Property.NodeScope, Setting.Property.Dynamic);
+        Settings nodeSettings = Settings.builder().put("cluster.setting", "initial_value").build();
+
+        ClusterSettings clusterSettings = new ClusterSettings(nodeSettings, Set.of(clusterSetting));
+        Settings newSettings = Settings.builder().put("cluster.setting", "updated_value").build();
+        clusterSettings.applySettings(newSettings);
+
+        // the value should be current when initializing the consumer
+        clusterSettings.initializeAndWatch(clusterSetting, value -> { assertThat(value, equalTo("updated_value")); });
+    }
+}