Added static setting for checkPendingFlushUpdate functionality of luc…

…ene writer

Signed-off-by: Prabhat Sharma <[email protected]>

CHANGELOG.md

@@ -116,6 +116,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Make search query counters dynamic to support all query types ([#12601](https://github.com/opensearch-project/OpenSearch/pull/12601))
 - [Tiered caching] Add policies controlling which values can enter pluggable caches [EXPERIMENTAL] ([#12542](https://github.com/opensearch-project/OpenSearch/pull/12542))
 - [Tiered caching] Add Stale keys Management and CacheCleaner to IndicesRequestCache ([#12625](https://github.com/opensearch-project/OpenSearch/pull/12625))
+- Introduce a new setting `index.check_pending_flush.enabled` to expose the ability to disable the check for pending flushes by write threads ([#12710](https://github.com/opensearch-project/OpenSearch/pull/12710))
 
 ### Dependencies
 - Bump `peter-evans/find-comment` from 2 to 3 ([#12288](https://github.com/opensearch-project/OpenSearch/pull/12288))

server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java

@@ -207,6 +207,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
                 IndexSettings.INDEX_MERGE_ON_FLUSH_MAX_FULL_FLUSH_MERGE_WAIT_TIME,
                 IndexSettings.INDEX_MERGE_ON_FLUSH_POLICY,
                 IndexSettings.INDEX_MERGE_POLICY,
+                IndexSettings.INDEX_CHECK_PENDING_FLUSH_ENABLED,
                 LogByteSizeMergePolicyProvider.INDEX_LBS_MERGE_POLICY_MERGE_FACTOR_SETTING,
                 LogByteSizeMergePolicyProvider.INDEX_LBS_MERGE_POLICY_MIN_MERGE_SETTING,
                 LogByteSizeMergePolicyProvider.INDEX_LBS_MAX_MERGE_SEGMENT_SETTING,

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

@@ -625,6 +625,13 @@ public static IndexMergePolicy fromString(String text) {
         Property.IndexScope
     );
 
+    public static final Setting<Boolean> INDEX_CHECK_PENDING_FLUSH_ENABLED = Setting.boolSetting(
+        "index.check_pending_flush.enabled",
+        true,
+        Property.IndexScope,
+        Property.Dynamic
+    );
+
     public static final Setting<String> TIME_SERIES_INDEX_MERGE_POLICY = Setting.simpleString(
         "indices.time_series_index.default_index_merge_policy",
         DEFAULT_POLICY,
@@ -817,7 +824,10 @@ private void setRetentionLeaseMillis(final TimeValue retentionLease) {
      * Specialized merge-on-flush policy if provided
      */
     private volatile UnaryOperator<MergePolicy> mergeOnFlushPolicy;
-
+    /**
+     * Is flush check by write threads enabled or not
+     */
+    private volatile boolean checkPendingFlushEnabled;
     /**
      * Is fuzzy set enabled for doc id
      */
@@ -960,6 +970,7 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         maxFullFlushMergeWaitTime = scopedSettings.get(INDEX_MERGE_ON_FLUSH_MAX_FULL_FLUSH_MERGE_WAIT_TIME);
         mergeOnFlushEnabled = scopedSettings.get(INDEX_MERGE_ON_FLUSH_ENABLED);
         setMergeOnFlushPolicy(scopedSettings.get(INDEX_MERGE_ON_FLUSH_POLICY));
+        checkPendingFlushEnabled = scopedSettings.get(INDEX_CHECK_PENDING_FLUSH_ENABLED);
         defaultSearchPipeline = scopedSettings.get(DEFAULT_SEARCH_PIPELINE);
         /* There was unintentional breaking change got introduced with [OpenSearch-6424](https://github.com/opensearch-project/OpenSearch/pull/6424) (version 2.7).
          * For indices created prior version (prior to 2.7) which has IndexSort type, they used to type cast the SortField.Type
@@ -1076,6 +1087,7 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         scopedSettings.addSettingsUpdateConsumer(INDEX_MERGE_ON_FLUSH_MAX_FULL_FLUSH_MERGE_WAIT_TIME, this::setMaxFullFlushMergeWaitTime);
         scopedSettings.addSettingsUpdateConsumer(INDEX_MERGE_ON_FLUSH_ENABLED, this::setMergeOnFlushEnabled);
         scopedSettings.addSettingsUpdateConsumer(INDEX_MERGE_ON_FLUSH_POLICY, this::setMergeOnFlushPolicy);
+        scopedSettings.addSettingsUpdateConsumer(INDEX_CHECK_PENDING_FLUSH_ENABLED, this::setCheckPendingFlushEnabled);
         scopedSettings.addSettingsUpdateConsumer(DEFAULT_SEARCH_PIPELINE, this::setDefaultSearchPipeline);
         scopedSettings.addSettingsUpdateConsumer(
             INDEX_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING,
@@ -1848,6 +1860,14 @@ private void setMergeOnFlushPolicy(String policy) {
         }
     }
 
+    private void setCheckPendingFlushEnabled(boolean enabled) {
+        this.checkPendingFlushEnabled = enabled;
+    }
+
+    public boolean isCheckPendingFlushEnabled() {
+        return checkPendingFlushEnabled;
+    }
+
     public Optional<UnaryOperator<MergePolicy>> getMergeOnFlushPolicy() {
         return Optional.ofNullable(mergeOnFlushPolicy);
     }

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

@@ -2336,6 +2336,7 @@ private IndexWriterConfig getIndexWriterConfig() {
             iwc.setMaxFullFlushMergeWaitMillis(0);
         }
 
+        iwc.setCheckPendingFlushUpdate(config().getIndexSettings().isCheckPendingFlushEnabled());
         iwc.setMergePolicy(new OpenSearchMergePolicy(mergePolicy));
         iwc.setSimilarity(engineConfig.getSimilarity());
         iwc.setRAMBufferSizeMB(engineConfig.getIndexingBufferSize().getMbFrac());