Archive unknown or invalid settings on updates (#28888)

Today we can end up in a situation where the cluster state contains
unknown or invalid settings. This can happen easily during a rolling
upgrade. For example, consider two nodes that are on a version that
considers the setting foo.bar to be known and valid. Assume one of these
nodes is restarted on a higher version that considers foo.bar to now be
either unknown or invalid, and then the second node is restarted
too. Now, both nodes will be on a version that consider foo.bar to be
unknown or invalid yet this setting will still be contained in the
cluster state. This means that if a cluster settings update is applied
and we validate the settings update with the existing settings then
validation will fail. In such a state, the offending setting can not
even be removed. This commit helps out with this situation by archiving
any settings that are unknown or invalid at the time that a settings
update is applied. This allows the setting update to go through, and the
archived settings can be removed at a later time.

server/src/main/java/org/elasticsearch/action/admin/cluster/settings/SettingsUpdater.java

@@ -19,13 +19,20 @@
 
 package org.elasticsearch.action.admin.cluster.settings;
 
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.apache.logging.log4j.util.Supplier;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlocks;
 import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 
+import java.util.Map;
+
 import static org.elasticsearch.cluster.ClusterState.builder;
+import static org.elasticsearch.common.settings.AbstractScopedSettings.ARCHIVED_SETTINGS_PREFIX;
 
 /**
  * Updates transient and persistent cluster state settings if there are any changes
@@ -48,15 +55,34 @@ final class SettingsUpdater {
         return persistentUpdates.build();
     }
 
-    synchronized ClusterState updateSettings(final ClusterState currentState, Settings transientToApply, Settings persistentToApply) {
+    synchronized ClusterState updateSettings(
+            final ClusterState currentState, final Settings transientToApply, final Settings persistentToApply, final Logger logger) {
         boolean changed = false;
-        Settings.Builder transientSettings = Settings.builder();
-        transientSettings.put(currentState.metaData().transientSettings());
-        changed |= clusterSettings.updateDynamicSettings(transientToApply, transientSettings, transientUpdates, "transient");
 
+        /*
+         * Our cluster state could have unknown or invalid settings that are known and valid in a previous version of Elasticsearch. We can
+         * end up in this situation during a rolling upgrade where the previous version will infect the current version of Elasticsearch
+         * with settings that the current version either no longer knows about or now considers to have invalid values. When the current
+         * version of Elasticsearch becomes infected with a cluster state containing such settings, we need to skip validating such settings
+         * and instead archive them. Consequently, for the current transient and persistent settings in the cluster state we do the
+         * following:
+         *  - split existing settings instance into two with the known and valid settings in one, and the unknown or invalid in another
+         *    (note that existing archived settings are included in the known and valid settings)
+         *  - validate the incoming settings update combined with the existing known and valid settings
+         *  - merge in the archived unknown or invalid settings
+         */
+        final Tuple<Settings, Settings> partitionedTransientSettings =
+                partitionKnownAndValidSettings(currentState.metaData().transientSettings(), "transient", logger);
+        final Settings knownAndValidTransientSettings = partitionedTransientSettings.v1();
+        final Settings unknownOrInvalidTransientSettings = partitionedTransientSettings.v2();
+        final Settings.Builder transientSettings = Settings.builder().put(knownAndValidTransientSettings);
+        changed |= clusterSettings.updateDynamicSettings(transientToApply, transientSettings, transientUpdates, "transient");
 
-        Settings.Builder persistentSettings = Settings.builder();
-        persistentSettings.put(currentState.metaData().persistentSettings());
+        final Tuple<Settings, Settings> partitionedPersistentSettings =
+                partitionKnownAndValidSettings(currentState.metaData().persistentSettings(), "persistent", logger);
+        final Settings knownAndValidPersistentSettings = partitionedPersistentSettings.v1();
+        final Settings unknownOrInvalidPersistentSettings = partitionedPersistentSettings.v2();
+        final Settings.Builder persistentSettings = Settings.builder().put(knownAndValidPersistentSettings);
         changed |= clusterSettings.updateDynamicSettings(persistentToApply, persistentSettings, persistentUpdates, "persistent");
 
         final ClusterState clusterState;
@@ -69,8 +95,8 @@ final class SettingsUpdater {
             clusterSettings.validate(persistentFinalSettings, true);
 
             MetaData.Builder metaData = MetaData.builder(currentState.metaData())
-                    .persistentSettings(persistentFinalSettings)
-                    .transientSettings(transientFinalSettings);
+                    .transientSettings(Settings.builder().put(transientFinalSettings).put(unknownOrInvalidTransientSettings).build())
+                    .persistentSettings(Settings.builder().put(persistentFinalSettings).put(unknownOrInvalidPersistentSettings).build());
 
             ClusterBlocks.Builder blocks = ClusterBlocks.builder().blocks(currentState.blocks());
             boolean updatedReadOnly = MetaData.SETTING_READ_ONLY_SETTING.get(metaData.persistentSettings())
@@ -102,5 +128,46 @@ final class SettingsUpdater {
         return clusterState;
     }
 
+    /**
+     * Partitions the settings into those that are known and valid versus those that are unknown or invalid. The resulting tuple contains
+     * the known and valid settings in the first component and the unknown or invalid settings in the second component. Note that archived
+     * settings contained in the settings to partition are included in the first component.
+     *
+     * @param settings     the settings to partition
+     * @param settingsType a string to identify the settings (for logging)
+     * @param logger       a logger to sending warnings to
+     * @return the partitioned settings
+     */
+    private Tuple<Settings, Settings> partitionKnownAndValidSettings(
+            final Settings settings, final String settingsType, final Logger logger) {
+        final Settings existingArchivedSettings = settings.filter(k -> k.startsWith(ARCHIVED_SETTINGS_PREFIX));
+        final Settings settingsExcludingExistingArchivedSettings =
+                settings.filter(k -> k.startsWith(ARCHIVED_SETTINGS_PREFIX) == false);
+        final Settings settingsWithUnknownOrInvalidArchived = clusterSettings.archiveUnknownOrInvalidSettings(
+                settingsExcludingExistingArchivedSettings,
+                e -> logUnknownSetting(settingsType, e, logger),
+                (e, ex) -> logInvalidSetting(settingsType, e, ex, logger));
+        return Tuple.tuple(
+                Settings.builder()
+                        .put(settingsWithUnknownOrInvalidArchived.filter(k -> k.startsWith(ARCHIVED_SETTINGS_PREFIX) == false))
+                        .put(existingArchivedSettings)
+                        .build(),
+                settingsWithUnknownOrInvalidArchived.filter(k -> k.startsWith(ARCHIVED_SETTINGS_PREFIX)));
+    }
+
+    private void logUnknownSetting(final String settingType, final Map.Entry<String, String> e, final Logger logger) {
+        logger.warn("ignoring existing unknown {} setting: [{}] with value [{}]; archiving", settingType, e.getKey(), e.getValue());
+    }
+
+    private void logInvalidSetting(
+            final String settingType, final Map.Entry<String, String> e, final IllegalArgumentException ex, final Logger logger) {
+        logger.warn(
+                (Supplier<?>)
+                        () -> new ParameterizedMessage("ignoring existing invalid {} setting: [{}] with value [{}]; archiving",
+                                settingType,
+                                e.getKey(),
+                                e.getValue()),
+                ex);
+    }
 
 }

server/src/main/java/org/elasticsearch/action/admin/cluster/settings/TransportClusterUpdateSettingsAction.java

@@ -180,7 +180,8 @@ public class TransportClusterUpdateSettingsAction extends
 
             @Override
             public ClusterState execute(final ClusterState currentState) {
-                ClusterState clusterState = updater.updateSettings(currentState, request.transientSettings(), request.persistentSettings());
+                ClusterState clusterState =
+                        updater.updateSettings(currentState, request.transientSettings(), request.persistentSettings(), logger);
                 changed = clusterState != currentState;
                 return clusterState;
             }

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

@@ -28,11 +28,20 @@ import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESTestCase;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiFunction;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.elasticsearch.common.settings.AbstractScopedSettings.ARCHIVED_SETTINGS_PREFIX;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.not;
+
 public class SettingsUpdaterTests extends ESTestCase {
 
 
@@ -51,7 +60,7 @@ public class SettingsUpdaterTests extends ESTestCase {
                 .put(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING.getKey(), 4.5).build());
         ClusterState build = builder.metaData(metaData).build();
         ClusterState clusterState = updater.updateSettings(build, Settings.builder().put(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.getKey(), 0.5).build(),
-            Settings.builder().put(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.getKey(), 0.4).build());
+            Settings.builder().put(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.getKey(), 0.4).build(), logger);
         assertNotSame(clusterState, build);
         assertEquals(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.get(clusterState.metaData().persistentSettings()), 0.4, 0.1);
         assertEquals(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING.get(clusterState.metaData().persistentSettings()), 2.5, 0.1);
@@ -59,14 +68,14 @@ public class SettingsUpdaterTests extends ESTestCase {
         assertEquals(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING.get(clusterState.metaData().transientSettings()), 4.5, 0.1);
 
         clusterState = updater.updateSettings(clusterState, Settings.builder().putNull("cluster.routing.*").build(),
-            Settings.EMPTY);
+            Settings.EMPTY, logger);
         assertEquals(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.get(clusterState.metaData().persistentSettings()), 0.4, 0.1);
         assertEquals(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING.get(clusterState.metaData().persistentSettings()), 2.5, 0.1);
         assertFalse(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.exists(clusterState.metaData().transientSettings()));
         assertFalse(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING.exists(clusterState.metaData().transientSettings()));
 
         clusterState = updater.updateSettings(clusterState,
-            Settings.EMPTY,  Settings.builder().putNull("cluster.routing.*").put(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.getKey(), 10.0).build());
+            Settings.EMPTY,  Settings.builder().putNull("cluster.routing.*").put(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.getKey(), 10.0).build(), logger);
 
         assertEquals(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.get(clusterState.metaData().persistentSettings()), 10.0, 0.1);
         assertFalse(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING.exists(clusterState.metaData().persistentSettings()));
@@ -93,7 +102,7 @@ public class SettingsUpdaterTests extends ESTestCase {
 
         try {
             updater.updateSettings(build, Settings.builder().put(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.getKey(), "not a float").build(),
-                Settings.builder().put(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.getKey(), "not a float").put(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING.getKey(), 1.0f).build());
+                Settings.builder().put(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.getKey(), "not a float").put(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING.getKey(), 1.0f).build(), logger);
             fail("all or nothing");
         } catch (IllegalArgumentException ex) {
             logger.info("", ex);
@@ -119,21 +128,21 @@ public class SettingsUpdaterTests extends ESTestCase {
         ClusterState build = builder.metaData(metaData).build();
 
         ClusterState clusterState = updater.updateSettings(build, Settings.builder().put(MetaData.SETTING_READ_ONLY_SETTING.getKey(), true).build(),
-            Settings.builder().put(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.getKey(), 1.6).put(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING.getKey(), 1.0f).build());
+            Settings.builder().put(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.getKey(), 1.6).put(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING.getKey(), 1.0f).build(), logger);
         assertEquals(clusterState.blocks().global().size(), 1);
         assertEquals(clusterState.blocks().global().iterator().next(), MetaData.CLUSTER_READ_ONLY_BLOCK);
 
         clusterState = updater.updateSettings(build, Settings.EMPTY,
-            Settings.builder().put(MetaData.SETTING_READ_ONLY_SETTING.getKey(), false).build());
+            Settings.builder().put(MetaData.SETTING_READ_ONLY_SETTING.getKey(), false).build(), logger);
         assertEquals(clusterState.blocks().global().size(), 0);
 
 
         clusterState = updater.updateSettings(build, Settings.builder().put(MetaData.SETTING_READ_ONLY_ALLOW_DELETE_SETTING.getKey(), true).build(),
-            Settings.builder().put(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.getKey(), 1.6).put(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING.getKey(), 1.0f).build());
+            Settings.builder().put(BalancedShardsAllocator.INDEX_BALANCE_FACTOR_SETTING.getKey(), 1.6).put(BalancedShardsAllocator.SHARD_BALANCE_FACTOR_SETTING.getKey(), 1.0f).build(), logger);
         assertEquals(clusterState.blocks().global().size(), 1);
         assertEquals(clusterState.blocks().global().iterator().next(), MetaData.CLUSTER_READ_ONLY_ALLOW_DELETE_BLOCK);
         clusterState = updater.updateSettings(build, Settings.EMPTY,
-            Settings.builder().put(MetaData.SETTING_READ_ONLY_ALLOW_DELETE_SETTING.getKey(), false).build());
+            Settings.builder().put(MetaData.SETTING_READ_ONLY_ALLOW_DELETE_SETTING.getKey(), false).build(), logger);
         assertEquals(clusterState.blocks().global().size(), 0);
 
     }
@@ -151,16 +160,317 @@ public class SettingsUpdaterTests extends ESTestCase {
                 ClusterState.builder(new ClusterName("foo")).metaData(MetaData.builder().persistentSettings(settings).build()).build();
 
         final Settings toApplyDebug = Settings.builder().put("logger.org.elasticsearch", "debug").build();
-        final ClusterState afterDebug = settingsUpdater.updateSettings(clusterState, toApplyDebug, Settings.EMPTY);
+        final ClusterState afterDebug = settingsUpdater.updateSettings(clusterState, toApplyDebug, Settings.EMPTY, logger);
         assertSettingDeprecationsAndWarnings(new Setting<?>[] { deprecatedSetting });
 
         final Settings toApplyUnset = Settings.builder().putNull("logger.org.elasticsearch").build();
-        final ClusterState afterUnset = settingsUpdater.updateSettings(afterDebug, toApplyUnset, Settings.EMPTY);
+        final ClusterState afterUnset = settingsUpdater.updateSettings(afterDebug, toApplyUnset, Settings.EMPTY, logger);
         assertSettingDeprecationsAndWarnings(new Setting<?>[] { deprecatedSetting });
 
         // we also check that if no settings are changed, deprecation logging still occurs
-        settingsUpdater.updateSettings(afterUnset, toApplyUnset, Settings.EMPTY);
+        settingsUpdater.updateSettings(afterUnset, toApplyUnset, Settings.EMPTY, logger);
         assertSettingDeprecationsAndWarnings(new Setting<?>[] { deprecatedSetting });
     }
 
+    public void testUpdateWithUnknownAndSettings() {
+        // we will randomly apply some new dynamic persistent and transient settings
+        final int numberOfDynamicSettings = randomIntBetween(1, 8);
+        final List<Setting<String>> dynamicSettings = new ArrayList<>(numberOfDynamicSettings);
+        for (int i = 0; i < numberOfDynamicSettings; i++) {
+            final Setting<String> dynamicSetting = Setting.simpleString("dynamic.setting" + i, Property.Dynamic, Property.NodeScope);
+            dynamicSettings.add(dynamicSetting);
+        }
+
+        // these are invalid settings that exist as either persistent or transient settings
+        final int numberOfInvalidSettings = randomIntBetween(0, 7);
+        final List<Setting<String>> invalidSettings = new ArrayList<>(numberOfInvalidSettings);
+        for (int i = 0; i < numberOfInvalidSettings; i++) {
+            final Setting<String> invalidSetting = Setting.simpleString(
+                    "invalid.setting" + i,
+                    (value, settings) -> {
+                        throw new IllegalArgumentException("invalid");
+                    },
+                    Property.NodeScope);
+            invalidSettings.add(invalidSetting);
+        }
+
+        // these are unknown settings that exist as either persistent or transient settings
+        final int numberOfUnknownSettings = randomIntBetween(0, 7);
+        final List<Setting<String>> unknownSettings = new ArrayList<>(numberOfUnknownSettings);
+        for (int i = 0; i < numberOfUnknownSettings; i++) {
+            final Setting<String> unknownSetting = Setting.simpleString("unknown.setting" + i, Property.NodeScope);
+            unknownSettings.add(unknownSetting);
+        }
+
+        final Settings.Builder existingPersistentSettings = Settings.builder();
+        final Settings.Builder existingTransientSettings = Settings.builder();
+
+        for (final Setting<String> dynamicSetting : dynamicSettings) {
+            switch (randomIntBetween(0, 2)) {
+                case 0:
+                    existingPersistentSettings.put(dynamicSetting.getKey(), "existing_value");
+                    break;
+                case 1:
+                    existingTransientSettings.put(dynamicSetting.getKey(), "existing_value");
+                    break;
+                case 2:
+                    break;
+            }
+        }
+
+        for (final Setting<String> invalidSetting : invalidSettings) {
+            if (randomBoolean()) {
+                existingPersistentSettings.put(invalidSetting.getKey(), "value");
+            } else {
+                existingTransientSettings.put(invalidSetting.getKey(), "value");
+            }
+        }
+
+        for (final Setting<String> unknownSetting : unknownSettings) {
+            if (randomBoolean()) {
+                existingPersistentSettings.put(unknownSetting.getKey(), "value");
+            } else {
+                existingTransientSettings.put(unknownSetting.getKey(), "value");
+            }
+        }
+
+        // register all the known settings (note that we do not register the unknown settings)
+        final Set<Setting<?>> knownSettings =
+                Stream.concat(
+                        ClusterSettings.BUILT_IN_CLUSTER_SETTINGS.stream(),
+                        Stream.concat(dynamicSettings.stream(), invalidSettings.stream()))
+                        .collect(Collectors.toSet());
+        final ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, knownSettings);
+        for (final Setting<String> dynamicSetting : dynamicSettings) {
+            clusterSettings.addSettingsUpdateConsumer(dynamicSetting, s -> {});
+        }
+        final SettingsUpdater settingsUpdater = new SettingsUpdater(clusterSettings);
+        final MetaData.Builder metaDataBuilder =
+                MetaData.builder()
+                        .persistentSettings(existingPersistentSettings.build())
+                        .transientSettings(existingTransientSettings.build());
+        final ClusterState clusterState = ClusterState.builder(new ClusterName("cluster")).metaData(metaDataBuilder).build();
+
+        // prepare the dynamic settings update
+        final Settings.Builder persistentToApply = Settings.builder();
+        final Settings.Builder transientToApply = Settings.builder();
+        for (final Setting<String> dynamicSetting : dynamicSettings) {
+            switch (randomIntBetween(0, 2)) {
+                case 0:
+                    persistentToApply.put(dynamicSetting.getKey(), "new_value");
+                    break;
+                case 1:
+                    transientToApply.put(dynamicSetting.getKey(), "new_value");
+                    break;
+                case 2:
+                    break;
+            }
+        }
+
+        if (transientToApply.keys().isEmpty() && persistentToApply.keys().isEmpty()) {
+            // force a settings update otherwise our assertions below will fail
+            if (randomBoolean()) {
+                persistentToApply.put(dynamicSettings.get(0).getKey(), "new_value");
+            } else {
+                transientToApply.put(dynamicSettings.get(0).getKey(), "new_value");
+            }
+        }
+
+        final ClusterState clusterStateAfterUpdate =
+                settingsUpdater.updateSettings(clusterState, transientToApply.build(), persistentToApply.build(), logger);
+
+        // the invalid settings should be archived and not present in non-archived form
+        for (final Setting<String> invalidSetting : invalidSettings) {
+            if (existingPersistentSettings.keys().contains(invalidSetting.getKey())) {
+                assertThat(
+                        clusterStateAfterUpdate.metaData().persistentSettings().keySet(),
+                        hasItem(ARCHIVED_SETTINGS_PREFIX + invalidSetting.getKey()));
+            } else {
+                assertThat(
+                        clusterStateAfterUpdate.metaData().transientSettings().keySet(),
+                        hasItem(ARCHIVED_SETTINGS_PREFIX + invalidSetting.getKey()));
+            }
+            assertThat(
+                    clusterStateAfterUpdate.metaData().persistentSettings().keySet(),
+                    not(hasItem(invalidSetting.getKey())));
+            assertThat(
+                    clusterStateAfterUpdate.metaData().transientSettings().keySet(),
+                    not(hasItem(invalidSetting.getKey())));
+        }
+
+        // the unknown settings should be archived and not present in non-archived form
+        for (final Setting<String> unknownSetting : unknownSettings) {
+            if (existingPersistentSettings.keys().contains(unknownSetting.getKey())) {
+                assertThat(
+                        clusterStateAfterUpdate.metaData().persistentSettings().keySet(),
+                        hasItem(ARCHIVED_SETTINGS_PREFIX + unknownSetting.getKey()));
+            } else {
+                assertThat(
+                        clusterStateAfterUpdate.metaData().transientSettings().keySet(),
+                        hasItem(ARCHIVED_SETTINGS_PREFIX + unknownSetting.getKey()));
+            }
+            assertThat(
+                    clusterStateAfterUpdate.metaData().persistentSettings().keySet(),
+                    not(hasItem(unknownSetting.getKey())));
+            assertThat(
+                    clusterStateAfterUpdate.metaData().transientSettings().keySet(),
+                    not(hasItem(unknownSetting.getKey())));
+        }
+
+        // the dynamic settings should be applied
+        for (final Setting<String> dynamicSetting : dynamicSettings) {
+            if (persistentToApply.keys().contains(dynamicSetting.getKey())) {
+                assertThat(clusterStateAfterUpdate.metaData().persistentSettings().keySet(), hasItem(dynamicSetting.getKey()));
+                assertThat(clusterStateAfterUpdate.metaData().persistentSettings().get(dynamicSetting.getKey()), equalTo("new_value"));
+            } else if (transientToApply.keys().contains(dynamicSetting.getKey())) {
+                assertThat(clusterStateAfterUpdate.metaData().transientSettings().keySet(), hasItem(dynamicSetting.getKey()));
+                assertThat(clusterStateAfterUpdate.metaData().transientSettings().get(dynamicSetting.getKey()), equalTo("new_value"));
+            } else {
+                if (existingPersistentSettings.keys().contains(dynamicSetting.getKey())) {
+                    assertThat(clusterStateAfterUpdate.metaData().persistentSettings().keySet(), hasItem(dynamicSetting.getKey()));
+                    assertThat(
+                            clusterStateAfterUpdate.metaData().persistentSettings().get(dynamicSetting.getKey()),
+                            equalTo("existing_value"));
+                } else if (existingTransientSettings.keys().contains(dynamicSetting.getKey())) {
+                    assertThat(clusterStateAfterUpdate.metaData().transientSettings().keySet(), hasItem(dynamicSetting.getKey()));
+                    assertThat(
+                            clusterStateAfterUpdate.metaData().transientSettings().get(dynamicSetting.getKey()),
+                            equalTo("existing_value"));
+                } else {
+                    assertThat(clusterStateAfterUpdate.metaData().persistentSettings().keySet(), not(hasItem(dynamicSetting.getKey())));
+                    assertThat(clusterStateAfterUpdate.metaData().transientSettings().keySet(), not(hasItem(dynamicSetting.getKey())));
+                }
+            }
+        }
+    }
+
+    public void testRemovingArchivedSettingsDoesNotRemoveNonArchivedInvalidOrUnknownSettings() {
+        // these are settings that are archived in the cluster state as either persistent or transient settings
+        final int numberOfArchivedSettings = randomIntBetween(1, 8);
+        final List<Setting<String>> archivedSettings = new ArrayList<>(numberOfArchivedSettings);
+        for (int i = 0; i < numberOfArchivedSettings; i++) {
+            final Setting<String> archivedSetting = Setting.simpleString("setting", Property.NodeScope);
+            archivedSettings.add(archivedSetting);
+        }
+
+        // these are invalid settings that exist as either persistent or transient settings
+        final int numberOfInvalidSettings = randomIntBetween(0, 7);
+        final List<Setting<String>> invalidSettings = new ArrayList<>(numberOfInvalidSettings);
+        for (int i = 0; i < numberOfInvalidSettings; i++) {
+            final Setting<String> invalidSetting = Setting.simpleString(
+                    "invalid.setting" + i,
+                    (value, settings) -> {
+                        throw new IllegalArgumentException("invalid");
+                    },
+                    Property.NodeScope);
+            invalidSettings.add(invalidSetting);
+        }
+
+        // these are unknown settings that exist as either persistent or transient settings
+        final int numberOfUnknownSettings = randomIntBetween(0, 7);
+        final List<Setting<String>> unknownSettings = new ArrayList<>(numberOfUnknownSettings);
+        for (int i = 0; i < numberOfUnknownSettings; i++) {
+            final Setting<String> unknownSetting = Setting.simpleString("unknown.setting" + i, Property.NodeScope);
+            unknownSettings.add(unknownSetting);
+        }
+
+        final Settings.Builder existingPersistentSettings = Settings.builder();
+        final Settings.Builder existingTransientSettings = Settings.builder();
+
+        for (final Setting<String> archivedSetting : archivedSettings) {
+            if (randomBoolean()) {
+                existingPersistentSettings.put(ARCHIVED_SETTINGS_PREFIX + archivedSetting.getKey(), "value");
+            } else {
+                existingTransientSettings.put(ARCHIVED_SETTINGS_PREFIX + archivedSetting.getKey(), "value");
+            }
+        }
+
+        for (final Setting<String> invalidSetting : invalidSettings) {
+            if (randomBoolean()) {
+                existingPersistentSettings.put(invalidSetting.getKey(), "value");
+            } else {
+                existingTransientSettings.put(invalidSetting.getKey(), "value");
+            }
+        }
+
+        for (final Setting<String> unknownSetting : unknownSettings) {
+            if (randomBoolean()) {
+                existingPersistentSettings.put(unknownSetting.getKey(), "value");
+            } else {
+                existingTransientSettings.put(unknownSetting.getKey(), "value");
+            }
+        }
+
+        // register all the known settings (not that we do not register the unknown settings)
+        final Set<Setting<?>> knownSettings =
+                Stream.concat(
+                        ClusterSettings.BUILT_IN_CLUSTER_SETTINGS.stream(),
+                        Stream.concat(archivedSettings.stream(), invalidSettings.stream()))
+                        .collect(Collectors.toSet());
+        final ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, knownSettings);
+        final SettingsUpdater settingsUpdater = new SettingsUpdater(clusterSettings);
+        final MetaData.Builder metaDataBuilder =
+                MetaData.builder()
+                        .persistentSettings(existingPersistentSettings.build())
+                        .transientSettings(existingTransientSettings.build());
+        final ClusterState clusterState = ClusterState.builder(new ClusterName("cluster")).metaData(metaDataBuilder).build();
+
+        final Settings.Builder persistentToApply = Settings.builder().put("archived.*", (String)null);
+        final Settings.Builder transientToApply = Settings.builder().put("archived.*", (String)null);
+
+        final ClusterState clusterStateAfterUpdate =
+                settingsUpdater.updateSettings(clusterState, transientToApply.build(), persistentToApply.build(), logger);
+
+        // existing archived settings are removed
+        for (final Setting<String> archivedSetting : archivedSettings) {
+            if (existingPersistentSettings.keys().contains(ARCHIVED_SETTINGS_PREFIX + archivedSetting.getKey())) {
+                assertThat(
+                        clusterStateAfterUpdate.metaData().persistentSettings().keySet(),
+                        not(hasItem(ARCHIVED_SETTINGS_PREFIX + archivedSetting.getKey())));
+            } else {
+                assertThat(
+                        clusterStateAfterUpdate.metaData().transientSettings().keySet(),
+                        not(hasItem(ARCHIVED_SETTINGS_PREFIX + archivedSetting.getKey())));
+            }
+        }
+
+        // the invalid settings should be archived and not present in non-archived form
+        for (final Setting<String> invalidSetting : invalidSettings) {
+            if (existingPersistentSettings.keys().contains(invalidSetting.getKey())) {
+                assertThat(
+                        clusterStateAfterUpdate.metaData().persistentSettings().keySet(),
+                        hasItem(ARCHIVED_SETTINGS_PREFIX + invalidSetting.getKey()));
+            } else {
+                assertThat(
+                        clusterStateAfterUpdate.metaData().transientSettings().keySet(),
+                        hasItem(ARCHIVED_SETTINGS_PREFIX + invalidSetting.getKey()));
+            }
+            assertThat(
+                    clusterStateAfterUpdate.metaData().persistentSettings().keySet(),
+                    not(hasItem(invalidSetting.getKey())));
+            assertThat(
+                    clusterStateAfterUpdate.metaData().transientSettings().keySet(),
+                    not(hasItem(invalidSetting.getKey())));
+        }
+
+        // the unknown settings should be archived and not present in non-archived form
+        for (final Setting<String> unknownSetting : unknownSettings) {
+            if (existingPersistentSettings.keys().contains(unknownSetting.getKey())) {
+                assertThat(
+                        clusterStateAfterUpdate.metaData().persistentSettings().keySet(),
+                        hasItem(ARCHIVED_SETTINGS_PREFIX + unknownSetting.getKey()));
+            } else {
+                assertThat(
+                        clusterStateAfterUpdate.metaData().transientSettings().keySet(),
+                        hasItem(ARCHIVED_SETTINGS_PREFIX + unknownSetting.getKey()));
+            }
+            assertThat(
+                    clusterStateAfterUpdate.metaData().persistentSettings().keySet(),
+                    not(hasItem(unknownSetting.getKey())));
+            assertThat(
+                    clusterStateAfterUpdate.metaData().transientSettings().keySet(),
+                    not(hasItem(unknownSetting.getKey())));
+        }
+    }
+
 }