Merge pull request #20383 from jimferenczi/max_merge_settings

Validate max thread/merge settings

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

@@ -263,8 +263,9 @@ public final class IndexSettings {
         scopedSettings.addSettingsUpdateConsumer(MergePolicyConfig.INDEX_MERGE_POLICY_MAX_MERGED_SEGMENT_SETTING, mergePolicyConfig::setMaxMergedSegment);
         scopedSettings.addSettingsUpdateConsumer(MergePolicyConfig.INDEX_MERGE_POLICY_SEGMENTS_PER_TIER_SETTING, mergePolicyConfig::setSegmentsPerTier);
         scopedSettings.addSettingsUpdateConsumer(MergePolicyConfig.INDEX_MERGE_POLICY_RECLAIM_DELETES_WEIGHT_SETTING, mergePolicyConfig::setReclaimDeletesWeight);
-        scopedSettings.addSettingsUpdateConsumer(MergeSchedulerConfig.MAX_THREAD_COUNT_SETTING, mergeSchedulerConfig::setMaxThreadCount);
+
-        scopedSettings.addSettingsUpdateConsumer(MergeSchedulerConfig.MAX_MERGE_COUNT_SETTING, mergeSchedulerConfig::setMaxMergeCount);
+        scopedSettings.addSettingsUpdateConsumer(MergeSchedulerConfig.MAX_THREAD_COUNT_SETTING, MergeSchedulerConfig.MAX_MERGE_COUNT_SETTING,
+            mergeSchedulerConfig::setMaxThreadAndMergeCount);
         scopedSettings.addSettingsUpdateConsumer(MergeSchedulerConfig.AUTO_THROTTLE_SETTING, mergeSchedulerConfig::setAutoThrottle);
         scopedSettings.addSettingsUpdateConsumer(INDEX_TRANSLOG_DURABILITY_SETTING, this::setTranslogDurability);
         scopedSettings.addSettingsUpdateConsumer(INDEX_TTL_DISABLE_PURGE_SETTING, this::setTTLPurgeDisabled);

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

@@ -69,13 +69,14 @@ public final class MergeSchedulerConfig {
     private volatile int maxMergeCount;
 
     MergeSchedulerConfig(IndexSettings indexSettings) {
-        maxThreadCount = indexSettings.getValue(MAX_THREAD_COUNT_SETTING);
+        setMaxThreadAndMergeCount(indexSettings.getValue(MAX_THREAD_COUNT_SETTING),
-        maxMergeCount = indexSettings.getValue(MAX_MERGE_COUNT_SETTING);
+            indexSettings.getValue(MAX_MERGE_COUNT_SETTING));
         this.autoThrottle = indexSettings.getValue(AUTO_THROTTLE_SETTING);
     }
 
     /**
      * Returns <code>true</code> iff auto throttle is enabled.
+     *
      * @see ConcurrentMergeScheduler#enableAutoIOThrottle()
      */
     public boolean isAutoThrottle() {
@@ -100,8 +101,19 @@ public final class MergeSchedulerConfig {
      * Expert: directly set the maximum number of merge threads and
      * simultaneous merges allowed.
      */
-    void setMaxThreadCount(int maxThreadCount) {
+    void setMaxThreadAndMergeCount(int maxThreadCount, int maxMergeCount) {
+        if (maxThreadCount < 1) {
+            throw new IllegalArgumentException("maxThreadCount should be at least 1");
+        }
+        if (maxMergeCount < 1) {
+            throw new IllegalArgumentException("maxMergeCount should be at least 1");
+        }
+        if (maxThreadCount > maxMergeCount) {
+            throw new IllegalArgumentException("maxThreadCount (= " + maxThreadCount +
+                ") should be <= maxMergeCount (= " + maxMergeCount + ")");
+        }
         this.maxThreadCount = maxThreadCount;
+        this.maxMergeCount = maxMergeCount;
     }
 
     /**
@@ -110,12 +122,4 @@ public final class MergeSchedulerConfig {
     public int getMaxMergeCount() {
         return maxMergeCount;
     }
-
-    /**
-     *
-     * Expert: set the maximum number of simultaneous merges allowed.
-     */
-    void setMaxMergeCount(int maxMergeCount) {
-        this.maxMergeCount = maxMergeCount;
-    }
 }

core/src/test/java/org/elasticsearch/indices/settings/UpdateSettingsIT.java

@@ -28,7 +28,9 @@ import org.apache.logging.log4j.core.filter.RegexFilter;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
 import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
 import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsResponse;
+import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
 import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
+import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequestBuilder;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.logging.Loggers;
@@ -417,6 +419,53 @@ public class UpdateSettingsIT extends ESIntegTestCase {
         }
     }
 
+    public void testInvalidMergeMaxThreadCount() throws IllegalAccessException {
+        CreateIndexRequestBuilder createBuilder = prepareCreate("test")
+            .setSettings(Settings.builder()
+                .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, "1")
+                .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, "0")
+                .put(MergePolicyConfig.INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE_SETTING.getKey(), "2")
+                .put(MergePolicyConfig.INDEX_MERGE_POLICY_SEGMENTS_PER_TIER_SETTING.getKey(), "2")
+                .put(MergeSchedulerConfig.MAX_THREAD_COUNT_SETTING.getKey(), "100")
+                .put(MergeSchedulerConfig.MAX_MERGE_COUNT_SETTING.getKey(), "10")
+            );
+        IllegalArgumentException exc = expectThrows(IllegalArgumentException.class,
+            () -> createBuilder.get());
+        assertThat(exc.getMessage(), equalTo("maxThreadCount (= 100) should be <= maxMergeCount (= 10)"));
+
+        assertAcked(prepareCreate("test")
+            .setSettings(Settings.builder()
+                .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, "1")
+                .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, "0")
+                .put(MergePolicyConfig.INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE_SETTING.getKey(), "2")
+                .put(MergePolicyConfig.INDEX_MERGE_POLICY_SEGMENTS_PER_TIER_SETTING.getKey(), "2")
+                .put(MergeSchedulerConfig.MAX_THREAD_COUNT_SETTING.getKey(), "100")
+                .put(MergeSchedulerConfig.MAX_MERGE_COUNT_SETTING.getKey(), "100")
+            ));
+
+        {
+            UpdateSettingsRequestBuilder updateBuilder = client().admin().indices()
+                .prepareUpdateSettings("test")
+                .setSettings(Settings.builder()
+                    .put(MergeSchedulerConfig.MAX_THREAD_COUNT_SETTING.getKey(), "1000")
+                );
+            exc = expectThrows(IllegalArgumentException.class,
+                () -> updateBuilder.get());
+            assertThat(exc.getMessage(), equalTo("maxThreadCount (= 1000) should be <= maxMergeCount (= 100)"));
+        }
+
+        {
+            UpdateSettingsRequestBuilder updateBuilder = client().admin().indices()
+                .prepareUpdateSettings("test")
+                .setSettings(Settings.builder()
+                    .put(MergeSchedulerConfig.MAX_MERGE_COUNT_SETTING.getKey(), "10")
+                );
+            exc = expectThrows(IllegalArgumentException.class,
+                () -> updateBuilder.get());
+            assertThat(exc.getMessage(), equalTo("maxThreadCount (= 100) should be <= maxMergeCount (= 10)"));
+        }
+    }
+
     // #6882: make sure we can change index.merge.scheduler.max_thread_count live
     public void testUpdateMergeMaxThreadCount() throws IllegalAccessException {
         MockAppender mockAppender = new MockAppender("testUpdateMergeMaxThreadCount");