make index.merge_on_flush.enabled, index.merge_on_flush.max_full_flush_merge_wait_time, index.merge_on_flush.policy, index.check_pending_flush.enabled dynamic

Signed-off-by: kkewwei <kewei.11@bytedance.com>
Signed-off-by: kkewwei <kkewwei@163.com>

Author: kkewwei

CHANGELOG.md

@@ -30,6 +30,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Convert transport-reactor-netty4 to use gradle version catalog [#17233](https://github.com/opensearch-project/OpenSearch/pull/17233)
 - Increase force merge threads to 1/8th of cores [#17255](https://github.com/opensearch-project/OpenSearch/pull/17255)
 - TieredSpilloverCache took-time threshold now guards heap tier as well as disk tier [#17190](https://github.com/opensearch-project/OpenSearch/pull/17190)
+- Make `index.merge_on_flush.enabled`, `index.merge_on_flush.max_full_flush_merge_wait_time`, `index.merge_on_flush.policy`, `index.check_pending_flush.enabled` dynamic
 
 ### Deprecated
 

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

@@ -648,7 +648,8 @@ public static IndexMergePolicy fromString(String text) {
     public static final Setting<Boolean> INDEX_CHECK_PENDING_FLUSH_ENABLED = Setting.boolSetting(
         "index.check_pending_flush.enabled",
         true,
-        Property.IndexScope
+        Property.IndexScope,
+        Property.Dynamic
     );
 
     public static final Setting<String> TIME_SERIES_INDEX_MERGE_POLICY = Setting.simpleString(
@@ -901,7 +902,7 @@ private void setRetentionLeaseMillis(final TimeValue retentionLease) {
     /**
      * Is flush check by write threads enabled or not
      */
-    private final boolean checkPendingFlushEnabled;
+    private volatile boolean checkPendingFlushEnabled;
     /**
      * Is fuzzy set enabled for doc id
      */
@@ -1200,6 +1201,11 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
             IndexMetadata.INDEX_REMOTE_TRANSLOG_REPOSITORY_SETTING,
             this::setRemoteStoreTranslogRepository
         );
+        scopedSettings.addSettingsUpdateConsumer(INDEX_CHECK_PENDING_FLUSH_ENABLED, this::setCheckPendingFlushEnabled);
+    }
+
+    public void setCheckPendingFlushEnabled(boolean checkPendingFlushEnabled) {
+        this.checkPendingFlushEnabled = checkPendingFlushEnabled;
     }
 
     private void setSearchIdleAfter(TimeValue searchIdleAfter) {

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

@@ -156,6 +156,7 @@ public class InternalEngine extends Engine {
     private volatile long lastDeleteVersionPruneTimeMSec;
 
     protected final TranslogManager translogManager;
+    private final IndexWriterConfig indexWriterConfig;
     protected final IndexWriter indexWriter;
     protected final LocalCheckpointTracker localCheckpointTracker;
     protected final AtomicLong maxUnsafeAutoIdTimestamp = new AtomicLong(-1);
@@ -166,6 +167,7 @@ public class InternalEngine extends Engine {
     protected final String historyUUID;
 
     private final OpenSearchConcurrentMergeScheduler mergeScheduler;
+    private MergePolicy unMergeOnFlushPolicy;
     private final ExternalReaderManager externalReaderManager;
     private final OpenSearchReaderManager internalReaderManager;
 
@@ -292,7 +294,8 @@ public void onFailure(String reason, Exception ex) {
                     translogManager::getLastSyncedGlobalCheckpoint
                 );
                 this.localCheckpointTracker = createLocalCheckpointTracker(localCheckpointTrackerSupplier);
-                writer = createWriter();
+                this.indexWriterConfig = getIndexWriterConfig();
+                writer = createWriter(indexWriterConfig);
                 bootstrapAppendOnlyInfoFromWriter(writer);
                 final Map<String, String> commitData = commitDataAsMap(writer);
                 historyUUID = loadHistoryUUID(commitData);
@@ -2303,9 +2306,8 @@ protected final ReferenceManager<OpenSearchDirectoryReader> getReferenceManager(
         }
     }
 
-    private IndexWriter createWriter() throws IOException {
+    private IndexWriter createWriter(IndexWriterConfig iwc) throws IOException {
         try {
-            final IndexWriterConfig iwc = getIndexWriterConfig();
             return createWriter(store.directory(), iwc);
         } catch (LockObtainFailedException ex) {
             logger.warn("could not lock IndexWriter", ex);
@@ -2336,26 +2338,27 @@ private IndexWriterConfig getIndexWriterConfig() {
         iwc.setMergeScheduler(mergeScheduler);
         // Give us the opportunity to upgrade old segments while performing
         // background merges
-        MergePolicy mergePolicy = config().getMergePolicy();
+        this.unMergeOnFlushPolicy = config().getMergePolicy();
         // always configure soft-deletes field so an engine with soft-deletes disabled can open a Lucene index with soft-deletes.
         iwc.setSoftDeletesField(Lucene.SOFT_DELETES_FIELD);
-        mergePolicy = new RecoverySourcePruneMergePolicy(
+        unMergeOnFlushPolicy = new RecoverySourcePruneMergePolicy(
             SourceFieldMapper.RECOVERY_SOURCE_NAME,
             softDeletesPolicy::getRetentionQuery,
             new SoftDeletesRetentionMergePolicy(
                 Lucene.SOFT_DELETES_FIELD,
                 softDeletesPolicy::getRetentionQuery,
-                new PrunePostingsMergePolicy(mergePolicy, IdFieldMapper.NAME)
+                new PrunePostingsMergePolicy(unMergeOnFlushPolicy, IdFieldMapper.NAME)
             )
         );
         boolean shuffleForcedMerge = Booleans.parseBoolean(System.getProperty("opensearch.shuffle_forced_merge", Boolean.TRUE.toString()));
         if (shuffleForcedMerge) {
             // We wrap the merge policy for all indices even though it is mostly useful for time-based indices
             // but there should be no overhead for other type of indices so it's simpler than adding a setting
             // to enable it.
-            mergePolicy = new ShuffleForcedMergePolicy(mergePolicy);
+            unMergeOnFlushPolicy = new ShuffleForcedMergePolicy(unMergeOnFlushPolicy);
         }
 
+        MergePolicy mergePolicy = unMergeOnFlushPolicy;
         if (config().getIndexSettings().isMergeOnFlushEnabled()) {
             final long maxFullFlushMergeWaitMillis = config().getIndexSettings().getMaxFullFlushMergeWaitTime().millis();
             if (maxFullFlushMergeWaitMillis > 0) {
@@ -2604,6 +2607,29 @@ public void onSettingsChanged(TimeValue translogRetentionAge, ByteSizeValue tran
             // the setting will be re-interpreted if it's set to true
             updateAutoIdTimestamp(Long.MAX_VALUE, true);
         }
+        IndexSettings indexSettings = engineConfig.getIndexSettings();
+        if (indexSettings.isCheckPendingFlushEnabled() != indexWriterConfig.isCheckPendingFlushOnUpdate()) {
+            indexWriterConfig.setCheckPendingFlushUpdate(indexSettings.isCheckPendingFlushEnabled());
+        }
+        if (indexSettings.isMergeOnFlushEnabled()) {
+            final long maxFullFlushMergeWaitMillis = indexSettings.getMaxFullFlushMergeWaitTime().millis();
+            if (maxFullFlushMergeWaitMillis > 0) {
+                indexWriterConfig.setMaxFullFlushMergeWaitMillis(maxFullFlushMergeWaitMillis);
+                final Optional<UnaryOperator<MergePolicy>> mergeOnFlushPolicy = indexSettings.getMergeOnFlushPolicy();
+                if (mergeOnFlushPolicy.isPresent()) {
+                    indexWriterConfig.setMergePolicy(new OpenSearchMergePolicy(mergeOnFlushPolicy.get().apply(unMergeOnFlushPolicy)));
+                }
+            } else {
+                logger.warn(
+                    "The {} is enabled but {} is set to 0, merge on flush will not be activated",
+                    IndexSettings.INDEX_MERGE_ON_FLUSH_ENABLED.getKey(),
+                    IndexSettings.INDEX_MERGE_ON_FLUSH_MAX_FULL_FLUSH_MERGE_WAIT_TIME.getKey()
+                );
+            }
+        } else {
+            indexWriterConfig.setMaxFullFlushMergeWaitMillis(0);
+            indexWriterConfig.setMergePolicy(new OpenSearchMergePolicy(unMergeOnFlushPolicy));
+        }
         final TranslogDeletionPolicy translogDeletionPolicy = translogManager.getDeletionPolicy();
         translogDeletionPolicy.setRetentionAgeInMillis(translogRetentionAge.millis());
         translogDeletionPolicy.setRetentionSizeInBytes(translogRetentionSize.getBytes());

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

@@ -67,6 +67,7 @@
 import org.apache.lucene.index.Terms;
 import org.apache.lucene.index.TermsEnum;
 import org.apache.lucene.index.TieredMergePolicy;
+import org.apache.lucene.sandbox.index.MergeOnFlushMergePolicy;
 import org.apache.lucene.search.IndexSearcher;
 import org.apache.lucene.search.MatchAllDocsQuery;
 import org.apache.lucene.search.ReferenceManager;
@@ -142,6 +143,7 @@
 import org.opensearch.index.seqno.RetentionLeases;
 import org.opensearch.index.seqno.SeqNoStats;
 import org.opensearch.index.seqno.SequenceNumbers;
+import org.opensearch.index.shard.OpenSearchMergePolicy;
 import org.opensearch.index.shard.ShardUtils;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.translog.DefaultTranslogDeletionPolicy;
@@ -6604,6 +6606,58 @@ public void testStressShouldPeriodicallyFlush() throws Exception {
         }
     }
 
+    public void testT1() throws Exception {
+        boolean checkPendingFlushEnabled = true;
+        boolean mergeOnFlushEnabled = true;
+        TimeValue maxFullFlushMergeWaitTime = TimeValue.timeValueSeconds(1);
+
+        final IndexSettings indexSettings = engine.config().getIndexSettings();
+        IndexMetadata indexMetadata = IndexMetadata.builder(indexSettings.getIndexMetadata())
+            .settings(
+                Settings.builder()
+                    .put(indexSettings.getSettings())
+                    .put(IndexSettings.INDEX_CHECK_PENDING_FLUSH_ENABLED.getKey(), checkPendingFlushEnabled)
+                    .put(IndexSettings.INDEX_MERGE_ON_FLUSH_ENABLED.getKey(),  mergeOnFlushEnabled)
+                    .put(IndexSettings.INDEX_MERGE_ON_FLUSH_MAX_FULL_FLUSH_MERGE_WAIT_TIME.getKey(), maxFullFlushMergeWaitTime)
+                    .put(IndexSettings.INDEX_MERGE_ON_FLUSH_POLICY.getKey(), "merge-on-flush")
+            )
+            .build();
+        indexSettings.updateIndexMetadata(indexMetadata);
+        engine.onSettingsChanged(
+            indexSettings.getTranslogRetentionAge(),
+            indexSettings.getTranslogRetentionSize(),
+            indexSettings.getSoftDeleteRetentionOperations()
+        );
+
+        assertEquals(checkPendingFlushEnabled, engine.getCurrentIndexWriterConfig().isCheckPendingFlushOnUpdate());
+        assertEquals(maxFullFlushMergeWaitTime.millis(), engine.getCurrentIndexWriterConfig().getMaxFullFlushMergeWaitMillis());
+        MergePolicy mergePolicy = engine.getCurrentIndexWriterConfig().getMergePolicy();
+        assertTrue(mergePolicy instanceof OpenSearchMergePolicy);
+        assertTrue(((OpenSearchMergePolicy) mergePolicy).getDelegate() instanceof MergeOnFlushMergePolicy);
+
+        mergeOnFlushEnabled = false;
+        indexMetadata = IndexMetadata.builder(indexSettings.getIndexMetadata())
+            .settings(
+                Settings.builder()
+                    .put(indexSettings.getSettings())
+                    .put(IndexSettings.INDEX_MERGE_ON_FLUSH_ENABLED.getKey(),  mergeOnFlushEnabled)
+                    .put(IndexSettings.INDEX_MERGE_ON_FLUSH_MAX_FULL_FLUSH_MERGE_WAIT_TIME.getKey(), maxFullFlushMergeWaitTime)
+                    .put(IndexSettings.INDEX_MERGE_ON_FLUSH_POLICY.getKey(), "merge-on-flush")
+            )
+            .build();
+        indexSettings.updateIndexMetadata(indexMetadata);
+        engine.onSettingsChanged(
+            indexSettings.getTranslogRetentionAge(),
+            indexSettings.getTranslogRetentionSize(),
+            indexSettings.getSoftDeleteRetentionOperations()
+        );
+
+        assertEquals(0, engine.getCurrentIndexWriterConfig().getMaxFullFlushMergeWaitMillis());
+        mergePolicy = engine.getCurrentIndexWriterConfig().getMergePolicy();
+        assertTrue(mergePolicy instanceof OpenSearchMergePolicy);
+        assertTrue(((OpenSearchMergePolicy) mergePolicy).getDelegate() instanceof ShuffleForcedMergePolicy);
+    }
+
     public void testStressUpdateSameDocWhileGettingIt() throws IOException, InterruptedException {
         final int iters = randomIntBetween(1, 15);
         for (int i = 0; i < iters; i++) {