Added test and changed nit naming

Signed-off-by: Harshita Kaushik <harycash@amazon.com>

Author: harycash

server/src/main/java/org/opensearch/index/IndexService.java

@@ -1241,7 +1241,7 @@ public void onDefaultMaxMergeAtOnceChanged(int newDefaultMaxMergeAtOnce) {
      *
      * @param maxForceMergeMBPerSec the updated cluster max force merge MB per second rate.
      */
-    public void onMaxForceMergeMBPerSecUpdate(double maxForceMergeMBPerSec) {
+    public void onClusterLevelForceMergeMBPerSecUpdate(double maxForceMergeMBPerSec) {
         if (!MergeSchedulerConfig.MAX_FORCE_MERGE_MB_PER_SEC_SETTING.exists(indexSettings.getSettings())) {
             indexSettings.getMergeSchedulerConfig().setMaxForceMergeMBPerSec(maxForceMergeMBPerSec);
         }

server/src/main/java/org/opensearch/index/IndexSettings.java

@@ -1180,7 +1180,7 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         scopedSettings.addSettingsUpdateConsumer(MergeSchedulerConfig.AUTO_THROTTLE_SETTING, mergeSchedulerConfig::setAutoThrottle);
         scopedSettings.addSettingsUpdateConsumer(
             MergeSchedulerConfig.MAX_FORCE_MERGE_MB_PER_SEC_SETTING,
-            mergeSchedulerConfig::setMaxForceMergeMBPerSec
+            value -> mergeSchedulerConfig.updateMaxForceMergeMBPerSec(this)
         );
         scopedSettings.addSettingsUpdateConsumer(INDEX_TRANSLOG_DURABILITY_SETTING, this::setTranslogDurability);
         scopedSettings.addSettingsUpdateConsumer(INDEX_TRANSLOG_SYNC_INTERVAL_SETTING, this::setTranslogSyncInterval);

server/src/main/java/org/opensearch/index/MergeSchedulerConfig.java

@@ -140,13 +140,7 @@ public final class MergeSchedulerConfig {
         int maxMerge = indexSettings.getValue(MAX_MERGE_COUNT_SETTING);
         setMaxThreadAndMergeCount(maxThread, maxMerge);
         this.autoThrottle = indexSettings.getValue(AUTO_THROTTLE_SETTING);
-
-        // Index setting takes precedence over cluster setting
-        if (MAX_FORCE_MERGE_MB_PER_SEC_SETTING.exists(indexSettings.getSettings())) {
-            this.maxForceMergeMBPerSec = indexSettings.getValue(MAX_FORCE_MERGE_MB_PER_SEC_SETTING);
-        } else {
-            this.maxForceMergeMBPerSec = CLUSTER_MAX_FORCE_MERGE_MB_PER_SEC_SETTING.get(indexSettings.getNodeSettings());
-        }
+        updateMaxForceMergeMBPerSec(indexSettings);
     }
 
     /**
@@ -214,4 +208,18 @@ public double getMaxForceMergeMBPerSec() {
     void setMaxForceMergeMBPerSec(double maxForceMergeMBPerSec) {
         this.maxForceMergeMBPerSec = maxForceMergeMBPerSec;
     }
+
+    /**
+     * Updates the maximum force merge rate based on index settings, with fallback to cluster settings.
+     * This method handles the case where an index-level setting is removed and should
+     * fall back to the cluster-level setting.
+     */
+    public void updateMaxForceMergeMBPerSec(IndexSettings indexSettings) {
+        boolean hasIndexSetting = MAX_FORCE_MERGE_MB_PER_SEC_SETTING.exists(indexSettings.getSettings());
+        if (hasIndexSetting) {
+            this.maxForceMergeMBPerSec = indexSettings.getValue(MAX_FORCE_MERGE_MB_PER_SEC_SETTING);
+        } else {
+            this.maxForceMergeMBPerSec = CLUSTER_MAX_FORCE_MERGE_MB_PER_SEC_SETTING.get(indexSettings.getNodeSettings());
+        }
+    }
 }

server/src/main/java/org/opensearch/index/engine/OpenSearchConcurrentMergeScheduler.java

@@ -39,7 +39,6 @@
 import org.opensearch.common.logging.Loggers;
 import org.opensearch.common.metrics.CounterMetric;
 import org.opensearch.common.metrics.MeanMetric;
-import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.util.concurrent.ConcurrentCollections;
 import org.opensearch.common.util.concurrent.OpenSearchExecutors;
@@ -64,7 +63,7 @@
 class OpenSearchConcurrentMergeScheduler extends ConcurrentMergeScheduler {
 
     protected final Logger logger;
-    private final Settings indexSettings;
+    private final IndexSettings indexSettings;
     private final ShardId shardId;
 
     private final MeanMetric totalMerges = new MeanMetric();
@@ -83,7 +82,7 @@ class OpenSearchConcurrentMergeScheduler extends ConcurrentMergeScheduler {
     OpenSearchConcurrentMergeScheduler(ShardId shardId, IndexSettings indexSettings) {
         this.config = indexSettings.getMergeSchedulerConfig();
         this.shardId = shardId;
-        this.indexSettings = indexSettings.getSettings();
+        this.indexSettings = indexSettings;
         this.logger = Loggers.getLogger(getClass(), shardId);
         refreshConfig();
     }
@@ -192,7 +191,10 @@ protected boolean maybeStall(MergeSource mergeSource) {
     protected MergeThread getMergeThread(MergeSource mergeSource, MergePolicy.OneMerge merge) throws IOException {
         MergeThread thread = super.getMergeThread(mergeSource, merge);
         thread.setName(
-            OpenSearchExecutors.threadName(indexSettings, "[" + shardId.getIndexName() + "][" + shardId.id() + "]: " + thread.getName())
+            OpenSearchExecutors.threadName(
+                indexSettings.getSettings(),
+                "[" + shardId.getIndexName() + "][" + shardId.id() + "]: " + thread.getName()
+            )
         );
         return thread;
     }
@@ -215,6 +217,9 @@ MergeStats stats() {
     }
 
     void refreshConfig() {
+        // Update the config with current index settings before using it
+        config.updateMaxForceMergeMBPerSec(indexSettings);
+
         if (this.getMaxMergeCount() != config.getMaxMergeCount() || this.getMaxThreadCount() != config.getMaxThreadCount()) {
             this.setMaxMergesAndThreads(config.getMaxMergeCount(), config.getMaxThreadCount());
         }

server/src/main/java/org/opensearch/indices/IndicesService.java

@@ -595,7 +595,7 @@ protected void closeInternal() {
         clusterService.getClusterSettings()
             .addSettingsUpdateConsumer(
                 MergeSchedulerConfig.CLUSTER_MAX_FORCE_MERGE_MB_PER_SEC_SETTING,
-                this::onMaxForceMergeMBPerSecUpdate
+                this::onClusterLevelForceMergeMBPerSecUpdate
             );
     }
 
@@ -707,10 +707,10 @@ private void onDefaultMaxMergeAtOnceUpdate(int newDefaultMaxMergeAtOnce) {
      *
      * @param maxForceMergeMBPerSec the updated cluster max force merge MB per second rate.
      */
-    private void onMaxForceMergeMBPerSecUpdate(double maxForceMergeMBPerSec) {
+    private void onClusterLevelForceMergeMBPerSecUpdate(double maxForceMergeMBPerSec) {
         for (Map.Entry<String, IndexService> entry : indices.entrySet()) {
             IndexService indexService = entry.getValue();
-            indexService.onMaxForceMergeMBPerSecUpdate(maxForceMergeMBPerSec);
+            indexService.onClusterLevelForceMergeMBPerSecUpdate(maxForceMergeMBPerSec);
         }
     }
 

server/src/test/java/org/opensearch/index/engine/MergeRateLimitingTests.java

@@ -166,4 +166,53 @@ public void testDynamicRateLimitUpdates() throws Exception {
             Loggers.setLevel(logger, (Level) null);
         }
     }
+
+    /**
+     * Test that when index-level setting is removed, scheduler falls back to cluster-level setting
+     */
+    public void testFallbackToClusterSettingWhenIndexSettingRemoved() throws Exception {
+        MockAppender mockAppender = new MockAppender("testFallbackToClusterSettingWhenIndexSettingRemoved");
+        mockAppender.start();
+        final Logger logger = LogManager.getLogger(OpenSearchConcurrentMergeScheduler.class);
+        Loggers.addAppender(logger, mockAppender);
+        Loggers.setLevel(logger, Level.INFO);
+
+        try {
+            Settings nodeSettings = Settings.builder().put(CLUSTER_MAX_FORCE_MERGE_MB_PER_SEC_SETTING.getKey(), "50.0").build();
+
+            // Start with index-level setting that overrides cluster setting
+            Settings.Builder builder = Settings.builder()
+                .put(SETTING_VERSION_CREATED, Version.CURRENT)
+                .put(SETTING_NUMBER_OF_SHARDS, "1")
+                .put(SETTING_NUMBER_OF_REPLICAS, "0")
+                .put(MAX_FORCE_MERGE_MB_PER_SEC_SETTING.getKey(), "25.0");
+
+            IndexSettings indexSettings = new IndexSettings(newIndexMeta("test_index", builder.build()), nodeSettings);
+            ShardId shardId = new ShardId("test_index", "test_uuid", 0);
+
+            OpenSearchConcurrentMergeScheduler scheduler = new OpenSearchConcurrentMergeScheduler(shardId, indexSettings);
+
+            // Should initially use index-level setting
+            assertThat(scheduler.getForceMergeMBPerSec(), equalTo(25.0));
+
+            // Remove the index-level setting by NOT setting MAX_FORCE_MERGE_MB_PER_SEC_SETTING - should trigger fallback to cluster setting
+            mockAppender.reset();
+            Settings.Builder newBuilder = Settings.builder()
+                .put(SETTING_VERSION_CREATED, Version.CURRENT)
+                .put(SETTING_NUMBER_OF_SHARDS, "1")
+                .put(SETTING_NUMBER_OF_REPLICAS, "0");
+
+            indexSettings.updateIndexMetadata(newIndexMeta("test_index", newBuilder.build()));
+            scheduler.refreshConfig();
+
+            // Should now use cluster-level setting
+            assertThat(scheduler.getForceMergeMBPerSec(), equalTo(50.0));
+            assertTrue("Should log rate limit update when falling back to cluster setting", mockAppender.sawRateLimitUpdate);
+
+        } finally {
+            Loggers.removeAppender(logger, mockAppender);
+            mockAppender.stop();
+            Loggers.setLevel(logger, (Level) null);
+        }
+    }
 }