Add settings updater for 2 affix settings (#33050)

Today we can only have non-affix settings updated and consumed _together_.
Yet, there are use-cases where two affix settings depend on each other which
makes using the hard without consuming updates together. Unfortunately, there is
not straight forward way to have N settings updated together in a type-safe way
having 2 still serves a large portion of use-cases.

Author: s1monw

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

@@ -199,7 +199,7 @@ public synchronized <T> void addSettingsUpdateConsumer(Setting<T> setting, Consu
      * Also automatically adds empty consumers for all settings in order to activate logging
      */
     public synchronized void addSettingsUpdateConsumer(Consumer<Settings> consumer, List<? extends Setting<?>> settings) {
-        addSettingsUpdater(Setting.groupedSettingsUpdater(consumer, logger, settings));
+        addSettingsUpdater(Setting.groupedSettingsUpdater(consumer, settings));
     }
 
     /**
@@ -208,11 +208,78 @@ public synchronized void addSettingsUpdateConsumer(Consumer<Settings> consumer,
      */
     public synchronized <T> void addAffixUpdateConsumer(Setting.AffixSetting<T> setting,  BiConsumer<String, T> consumer,
                                                         BiConsumer<String, T> validator) {
+        ensureSettingIsRegistered(setting);
+        addSettingsUpdater(setting.newAffixUpdater(consumer, logger, validator));
+    }
+
+    /**
+     * Adds a affix settings consumer that accepts the values for two settings. The consumer is only notified if one or both settings change
+     * and if the provided validator succeeded.
+     * <p>
+     * Note: Only settings registered in {@link SettingsModule} can be changed dynamically.
+     * </p>
+     * This method registers a compound updater that is useful if two settings are depending on each other.
+     * The consumer is always provided with both values even if only one of the two changes.
+     */
+    public synchronized <A,B> void addAffixUpdateConsumer(Setting.AffixSetting<A> settingA, Setting.AffixSetting<B> settingB,
+                                                          BiConsumer<String, Tuple<A, B>> consumer,
+                                                          BiConsumer<String, Tuple<A, B>> validator) {
+        // it would be awesome to have a generic way to do that ie. a set of settings that map to an object with a builder
+        // down the road this would be nice to have!
+        ensureSettingIsRegistered(settingA);
+        ensureSettingIsRegistered(settingB);
+        SettingUpdater<Map<SettingUpdater<A>, A>> affixUpdaterA = settingA.newAffixUpdater((a,b)-> {}, logger, (a,b)-> {});
+        SettingUpdater<Map<SettingUpdater<B>, B>> affixUpdaterB = settingB.newAffixUpdater((a,b)-> {}, logger, (a,b)-> {});
+
+        addSettingsUpdater(new SettingUpdater<Map<String, Tuple<A, B>>>() {
+
+            @Override
+            public boolean hasChanged(Settings current, Settings previous) {
+                return affixUpdaterA.hasChanged(current, previous) || affixUpdaterB.hasChanged(current, previous);
+            }
+
+            @Override
+            public Map<String, Tuple<A, B>> getValue(Settings current, Settings previous) {
+                Map<String, Tuple<A, B>> map = new HashMap<>();
+                BiConsumer<String, A> aConsumer = (key, value) -> {
+                    assert map.containsKey(key) == false : "duplicate key: " + key;
+                    map.put(key, new Tuple<>(value, settingB.getConcreteSettingForNamespace(key).get(current)));
+                };
+                BiConsumer<String, B> bConsumer = (key, value) -> {
+                    Tuple<A, B> abTuple = map.get(key);
+                    if (abTuple != null) {
+                        map.put(key, new Tuple<>(abTuple.v1(), value));
+                    } else {
+                        assert settingA.getConcreteSettingForNamespace(key).get(current).equals(settingA.getConcreteSettingForNamespace
+                            (key).get(previous)) : "expected: " + settingA.getConcreteSettingForNamespace(key).get(current)
+                            + " but was " + settingA.getConcreteSettingForNamespace(key).get(previous);
+                        map.put(key, new Tuple<>(settingA.getConcreteSettingForNamespace(key).get(current), value));
+                    }
+                };
+                SettingUpdater<Map<SettingUpdater<A>, A>> affixUpdaterA = settingA.newAffixUpdater(aConsumer, logger, (a,b) ->{});
+                SettingUpdater<Map<SettingUpdater<B>, B>> affixUpdaterB = settingB.newAffixUpdater(bConsumer, logger, (a,b) ->{});
+                affixUpdaterA.apply(current, previous);
+                affixUpdaterB.apply(current, previous);
+                for (Map.Entry<String, Tuple<A, B>> entry : map.entrySet()) {
+                    validator.accept(entry.getKey(), entry.getValue());
+                }
+                return Collections.unmodifiableMap(map);
+            }
+
+            @Override
+            public void apply(Map<String, Tuple<A, B>> values, Settings current, Settings previous) {
+                for (Map.Entry<String, Tuple<A, B>> entry : values.entrySet()) {
+                    consumer.accept(entry.getKey(), entry.getValue());
+                }
+            }
+        });
+    }
+
+    private void ensureSettingIsRegistered(Setting.AffixSetting<?> setting) {
         final Setting<?> registeredSetting = this.complexMatchers.get(setting.getKey());
         if (setting != registeredSetting) {
             throw new IllegalArgumentException("Setting is not registered for key [" + setting.getKey() + "]");
         }
-        addSettingsUpdater(setting.newAffixUpdater(consumer, logger, validator));
     }
 
     /**

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

@@ -547,7 +547,7 @@ public String toString() {
         };
     }
 
-    static AbstractScopedSettings.SettingUpdater<Settings> groupedSettingsUpdater(Consumer<Settings> consumer, Logger logger,
+    static AbstractScopedSettings.SettingUpdater<Settings> groupedSettingsUpdater(Consumer<Settings> consumer,
                                                                                   final List<? extends Setting<?>> configuredSettings) {
 
         return new AbstractScopedSettings.SettingUpdater<Settings>() {

server/src/test/java/org/elasticsearch/common/settings/ScopedSettingsTests.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.routing.allocation.decider.FilterAllocationDecider;
 import org.elasticsearch.cluster.routing.allocation.decider.ShardsLimitAllocationDecider;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.logging.ESLoggerFactory;
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.Setting.Property;
@@ -180,6 +181,99 @@ public void testDependentSettings() {
         service.validate(Settings.builder().put("foo.test.bar", 7).build(), false);
     }
 
+    public void testTupleAffixUpdateConsumer() {
+        String prefix = randomAlphaOfLength(3) + "foo.";
+        String intSuffix = randomAlphaOfLength(3);
+        String listSuffix = randomAlphaOfLength(4);
+        Setting.AffixSetting<Integer> intSetting = Setting.affixKeySetting(prefix, intSuffix,
+            (k) ->  Setting.intSetting(k, 1, Property.Dynamic, Property.NodeScope));
+        Setting.AffixSetting<List<Integer>> listSetting = Setting.affixKeySetting(prefix, listSuffix,
+            (k) -> Setting.listSetting(k, Arrays.asList("1"), Integer::parseInt, Property.Dynamic, Property.NodeScope));
+        AbstractScopedSettings service = new ClusterSettings(Settings.EMPTY,new HashSet<>(Arrays.asList(intSetting, listSetting)));
+        Map<String, Tuple<List<Integer>, Integer>> results = new HashMap<>();
+        Function<String, String> listBuilder = g -> (prefix + g + "." + listSuffix);
+        Function<String, String> intBuilder = g -> (prefix + g + "." + intSuffix);
+        String group1 = randomAlphaOfLength(3);
+        String group2 = randomAlphaOfLength(4);
+        String group3 = randomAlphaOfLength(5);
+        BiConsumer<String, Tuple<List<Integer>, Integer>> listConsumer = results::put;
+
+        service.addAffixUpdateConsumer(listSetting, intSetting, listConsumer, (s, k) -> {
+            if (k.v1().isEmpty() && k.v2() == 2) {
+                throw new IllegalArgumentException("boom");
+            }
+        });
+        assertEquals(0, results.size());
+        service.applySettings(Settings.builder()
+            .put(intBuilder.apply(group1), 2)
+            .put(intBuilder.apply(group2), 7)
+            .putList(listBuilder.apply(group1), "16", "17")
+            .putList(listBuilder.apply(group2), "18", "19", "20")
+            .build());
+        assertEquals(2, results.get(group1).v2().intValue());
+        assertEquals(7, results.get(group2).v2().intValue());
+        assertEquals(Arrays.asList(16, 17), results.get(group1).v1());
+        assertEquals(Arrays.asList(18, 19, 20), results.get(group2).v1());
+        assertEquals(2, results.size());
+
+        results.clear();
+
+        service.applySettings(Settings.builder()
+            .put(intBuilder.apply(group1), 2)
+            .put(intBuilder.apply(group2), 7)
+            .putList(listBuilder.apply(group1), "16", "17")
+            .putNull(listBuilder.apply(group2)) // removed
+            .build());
+
+        assertNull(group1 + " wasn't changed", results.get(group1));
+        assertEquals(1, results.get(group2).v1().size());
+        assertEquals(Arrays.asList(1), results.get(group2).v1());
+        assertEquals(7, results.get(group2).v2().intValue());
+        assertEquals(1, results.size());
+        results.clear();
+
+        service.applySettings(Settings.builder()
+            .put(intBuilder.apply(group1), 2)
+            .put(intBuilder.apply(group2), 7)
+            .putList(listBuilder.apply(group1), "16", "17")
+            .putList(listBuilder.apply(group3), "5", "6") // added
+            .build());
+        assertNull(group1 + " wasn't changed", results.get(group1));
+        assertNull(group2 + " wasn't changed", results.get(group2));
+
+        assertEquals(2, results.get(group3).v1().size());
+        assertEquals(Arrays.asList(5, 6), results.get(group3).v1());
+        assertEquals(1, results.get(group3).v2().intValue());
+        assertEquals(1, results.size());
+        results.clear();
+
+        service.applySettings(Settings.builder()
+            .put(intBuilder.apply(group1), 4) // modified
+            .put(intBuilder.apply(group2), 7)
+            .putList(listBuilder.apply(group1), "16", "17")
+            .putList(listBuilder.apply(group3), "5", "6")
+            .build());
+        assertNull(group2 + " wasn't changed", results.get(group2));
+        assertNull(group3 + " wasn't changed", results.get(group3));
+
+        assertEquals(2, results.get(group1).v1().size());
+        assertEquals(Arrays.asList(16, 17), results.get(group1).v1());
+        assertEquals(4, results.get(group1).v2().intValue());
+        assertEquals(1, results.size());
+        results.clear();
+
+        IllegalArgumentException iae = expectThrows(IllegalArgumentException.class, () ->
+            service.applySettings(Settings.builder()
+                .put(intBuilder.apply(group1), 2) // modified to trip validator
+                .put(intBuilder.apply(group2), 7)
+                .putList(listBuilder.apply(group1)) // modified to trip validator
+                .putList(listBuilder.apply(group3), "5", "6")
+                .build())
+        );
+        assertEquals("boom", iae.getMessage());
+        assertEquals(0, results.size());
+    }
+
     public void testAddConsumerAffix() {
         Setting.AffixSetting<Integer> intSetting = Setting.affixKeySetting("foo.", "bar",
             (k) ->  Setting.intSetting(k, 1, Property.Dynamic, Property.NodeScope));
@@ -893,7 +987,7 @@ public void testInternalIndexSettingsFailsValidation() {
 
     public void testInternalIndexSettingsSkipValidation() {
         final Setting<String> internalIndexSetting = Setting.simpleString("index.internal", Property.InternalIndex, Property.IndexScope);
-        final IndexScopedSettings indexScopedSettings = 
+        final IndexScopedSettings indexScopedSettings =
                 new IndexScopedSettings(Settings.EMPTY, Collections.singleton(internalIndexSetting));
         // nothing should happen, validation should not throw an exception
         final Settings settings = Settings.builder().put("index.internal", "internal").build();

server/src/test/java/org/elasticsearch/common/settings/SettingTests.java

@@ -757,7 +757,7 @@ public void testTimeValue() {
     public void testSettingsGroupUpdater() {
         Setting<Integer> intSetting = Setting.intSetting("prefix.foo", 1, Property.NodeScope, Property.Dynamic);
         Setting<Integer> intSetting2 = Setting.intSetting("prefix.same", 1, Property.NodeScope, Property.Dynamic);
-        AbstractScopedSettings.SettingUpdater<Settings> updater = Setting.groupedSettingsUpdater(s -> {}, logger,
+        AbstractScopedSettings.SettingUpdater<Settings> updater = Setting.groupedSettingsUpdater(s -> {},
             Arrays.asList(intSetting, intSetting2));
 
         Settings current = Settings.builder().put("prefix.foo", 123).put("prefix.same", 5555).build();
@@ -768,7 +768,7 @@ public void testSettingsGroupUpdater() {
     public void testSettingsGroupUpdaterRemoval() {
         Setting<Integer> intSetting = Setting.intSetting("prefix.foo", 1, Property.NodeScope, Property.Dynamic);
         Setting<Integer> intSetting2 = Setting.intSetting("prefix.same", 1, Property.NodeScope, Property.Dynamic);
-        AbstractScopedSettings.SettingUpdater<Settings> updater = Setting.groupedSettingsUpdater(s -> {}, logger,
+        AbstractScopedSettings.SettingUpdater<Settings> updater = Setting.groupedSettingsUpdater(s -> {},
             Arrays.asList(intSetting, intSetting2));
 
         Settings current = Settings.builder().put("prefix.same", 5555).build();
@@ -783,7 +783,7 @@ public void testSettingsGroupUpdaterWithAffixSetting() {
         Setting.AffixSetting<String> affixSetting =
             Setting.affixKeySetting("prefix.foo.", "suffix", key -> Setting.simpleString(key,Property.NodeScope, Property.Dynamic));
 
-        AbstractScopedSettings.SettingUpdater<Settings> updater = Setting.groupedSettingsUpdater(s -> {}, logger,
+        AbstractScopedSettings.SettingUpdater<Settings> updater = Setting.groupedSettingsUpdater(s -> {},
             Arrays.asList(intSetting, prefixKeySetting, affixSetting));
 
         Settings.Builder currentSettingsBuilder = Settings.builder()