Add validation of index settings passed via cmd and elasticsearch.yaml

From this commit on we also validate all settings starting with `index.` that are
node-level settings configured in yaml files or via commandline arguments or system properties.
This check happens on node startup before the actual node is started.

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

@@ -179,6 +179,25 @@ public abstract class AbstractScopedSettings extends AbstractComponent {
        addSettingsUpdateConsumer(setting, consumer, (s) -> {});
     }
 
+    /**
+     * Validates that all settings in the builder are registered and valid
+     */
+    public final void validate(Settings.Builder settingsBuilder) {
+        for (Map.Entry<String, String> entry : settingsBuilder.internalMap().entrySet()) {
+            validate(entry.getKey(), entry.getValue());
+        }
+    }
+
+    /**
+     * * Validates that all given settings are registered and valid
+     */
+    public final void validate(Settings settings) {
+        for (Map.Entry<String, String> entry : settings.getAsMap().entrySet()) {
+            validate(entry.getKey(), entry.getValue());
+        }
+    }
+
+
     /**
      * Validates that the setting is valid
      */

core/src/main/java/org/elasticsearch/common/settings/IndexScopeSettings.java

@@ -69,12 +69,15 @@ import java.util.Collections;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Predicate;
 
 /**
  * Encapsulates all valid cluster level settings.
  */
 public final class IndexScopeSettings extends AbstractScopedSettings {
 
+    public static final Predicate<String> INDEX_SETTINGS_KEY_PREDICATE = (s) -> s.startsWith(IndexMetaData.INDEX_SETTING_PREFIX);
+
     public static Set<Setting<?>> BUILT_IN_INDEX_SETTINGS = Collections.unmodifiableSet(new HashSet<>(Arrays.asList(
         IndexSettings.INDEX_TTL_DISABLE_PURGE_SETTING,
         IndexStore.INDEX_STORE_THROTTLE_TYPE_SETTING,

core/src/main/java/org/elasticsearch/common/settings/Settings.java

@@ -58,6 +58,7 @@ import java.util.Set;
 import java.util.SortedMap;
 import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -212,6 +213,19 @@ public final class Settings implements ToXContent {
         return builder.build();
     }
 
+    /**
+     * Returns a new settings object that contains all setting of the current one filtered by the given settings key predicate.
+     */
+    public Settings filter(Predicate<String> predicate) {
+        Builder builder = new Builder();
+        for (Map.Entry<String, String> entry : getAsMap().entrySet()) {
+            if (predicate.test(entry.getKey())) {
+                builder.put(entry.getKey(), entry.getValue());
+            }
+        }
+        return builder.build();
+    }
+
     /**
      * Returns the settings mapped to the given setting name.
      */

core/src/main/java/org/elasticsearch/common/settings/SettingsModule.java

@@ -51,10 +51,13 @@ public class SettingsModule extends AbstractModule {
 
     @Override
     protected void configure() {
+        final IndexScopeSettings indexScopeSettings = new IndexScopeSettings(settings, new HashSet<>(this.indexSettings.values()));
+        // by now we are fully configured, lets check node level settings for unregistered index settings
+        indexScopeSettings.validate(settings.filter(IndexScopeSettings.INDEX_SETTINGS_KEY_PREDICATE));
         bind(Settings.class).toInstance(settings);
         bind(SettingsFilter.class).toInstance(settingsFilter);
         final ClusterSettings clusterSettings = new ClusterSettings(settings, new HashSet<>(this.clusterSettings.values()));
-        final IndexScopeSettings indexScopeSettings = new IndexScopeSettings(settings, new HashSet<>(this.indexSettings.values()));
+
         bind(ClusterSettings.class).toInstance(clusterSettings);
         bind(IndexScopeSettings.class).toInstance(indexScopeSettings);
     }

core/src/main/java/org/elasticsearch/index/IndexSettings.java

@@ -347,7 +347,7 @@ public final class IndexSettings {
         }
         this.indexMetaData = indexMetaData;
         final Settings existingSettings = this.settings;
-        if (existingSettings.getByPrefix(IndexMetaData.INDEX_SETTING_PREFIX).getAsMap().equals(newSettings.getByPrefix(IndexMetaData.INDEX_SETTING_PREFIX).getAsMap())) {
+        if (existingSettings.filter(IndexScopeSettings.INDEX_SETTINGS_KEY_PREDICATE).getAsMap().equals(newSettings.filter(IndexScopeSettings.INDEX_SETTINGS_KEY_PREDICATE).getAsMap())) {
             // nothing to update, same settings
             return false;
         }

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

@@ -179,6 +179,41 @@ public class ScopedSettingsTests extends ESTestCase {
         assertEquals("boom", copy.get(IndexModule.INDEX_STORE_TYPE_SETTING)); // test fallback to node settings
     }
 
+    public void testValidate() {
+        IndexScopeSettings settings = new IndexScopeSettings(
+            Settings.EMPTY,
+            IndexScopeSettings.BUILT_IN_INDEX_SETTINGS);
+        settings.validate(Settings.builder().put("index.store.type", "boom"));
+        settings.validate(Settings.builder().put("index.store.type", "boom").build());
+        try {
+            settings.validate(Settings.builder().put("index.store.type", "boom", "i.am.not.a.setting", true));
+            fail();
+        } catch (IllegalArgumentException e) {
+            assertEquals("unknown setting [i.am.not.a.setting]", e.getMessage());
+        }
+
+        try {
+            settings.validate(Settings.builder().put("index.store.type", "boom", "i.am.not.a.setting", true).build());
+            fail();
+        } catch (IllegalArgumentException e) {
+            assertEquals("unknown setting [i.am.not.a.setting]", e.getMessage());
+        }
+
+        try {
+            settings.validate(Settings.builder().put("index.store.type", "boom", "index.number_of_replicas", true).build());
+            fail();
+        } catch (IllegalArgumentException e) {
+            assertEquals("Failed to parse value [true] for setting [index.number_of_replicas]", e.getMessage());
+        }
+
+        try {
+            settings.validate("index.number_of_replicas", "true");
+            fail();
+        } catch (IllegalArgumentException e) {
+            assertEquals("Failed to parse value [true] for setting [index.number_of_replicas]", e.getMessage());
+        }
+    }
+
 
     public static IndexMetaData newIndexMeta(String name, Settings indexSettings) {
         Settings build = Settings.settingsBuilder().put(IndexMetaData.SETTING_VERSION_CREATED, Version.CURRENT)