Use feature flagged settings map

Signed-off-by: Jay Deng <jayd0104@gmail.com>

Author: jed326

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

@@ -360,7 +360,6 @@ public void apply(Settings value, Settings current, Settings previous) {
                 ClusterManagerService.CLUSTER_MANAGER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING,
                 SearchService.DEFAULT_SEARCH_TIMEOUT_SETTING,
                 SearchService.DEFAULT_ALLOW_PARTIAL_SEARCH_RESULTS,
-                SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING,
                 TransportSearchAction.SHARD_COUNT_LIMIT_SETTING,
                 TransportSearchAction.SEARCH_CANCEL_AFTER_TIME_INTERVAL_SETTING,
                 RemoteClusterService.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
@@ -669,6 +668,8 @@ public void apply(Settings value, Settings current, Settings previous) {
             IndicesService.CLUSTER_REMOTE_STORE_REPOSITORY_SETTING,
             IndicesService.CLUSTER_REMOTE_TRANSLOG_STORE_ENABLED_SETTING,
             IndicesService.CLUSTER_REMOTE_TRANSLOG_REPOSITORY_SETTING
-        )
+        ),
+        List.of(FeatureFlags.CONCURRENT_SEGMENT_SEARCH),
+        List.of(SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING)
     );
 }

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

@@ -198,7 +198,6 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
                 IndexSettings.INDEX_MERGE_ON_FLUSH_MAX_FULL_FLUSH_MERGE_WAIT_TIME,
                 IndexSettings.INDEX_MERGE_ON_FLUSH_POLICY,
                 IndexSettings.DEFAULT_SEARCH_PIPELINE,
-                IndexSettings.INDEX_CONCURRENT_SEGMENT_SEARCH_SETTING,
 
                 // Settings for Searchable Snapshots
                 IndexSettings.SEARCHABLE_SNAPSHOT_REPOSITORY,
@@ -236,7 +235,9 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
             IndexMetadata.INDEX_REMOTE_TRANSLOG_STORE_ENABLED_SETTING,
             IndexMetadata.INDEX_REMOTE_TRANSLOG_REPOSITORY_SETTING,
             IndexMetadata.INDEX_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING
-        )
+        ),
+        FeatureFlags.CONCURRENT_SEGMENT_SEARCH,
+        List.of(IndexSettings.INDEX_CONCURRENT_SEGMENT_SEARCH_SETTING)
     );
 
     public static final IndexScopedSettings DEFAULT_SCOPED_SETTINGS = new IndexScopedSettings(Settings.EMPTY, BUILT_IN_INDEX_SETTINGS);

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

@@ -62,7 +62,6 @@
 import java.util.function.Function;
 import java.util.function.UnaryOperator;
 
-import static org.opensearch.common.util.FeatureFlags.CONCURRENT_SEGMENT_SEARCH;
 import static org.opensearch.common.util.FeatureFlags.SEARCHABLE_SNAPSHOT_EXTENDED_COMPATIBILITY;
 import static org.opensearch.common.util.FeatureFlags.SEARCH_PIPELINE;
 import static org.opensearch.index.mapper.MapperService.INDEX_MAPPING_DEPTH_LIMIT_SETTING;
@@ -673,7 +672,6 @@ private void setRetentionLeaseMillis(final TimeValue retentionLease) {
     private volatile long mappingTotalFieldsLimit;
     private volatile long mappingDepthLimit;
     private volatile long mappingFieldNameLengthLimit;
-    private volatile boolean indexConcurrentSegmentSearchEnabled;
 
     /**
      * The maximum number of refresh listeners allows on this shard.
@@ -838,7 +836,6 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         mergeOnFlushEnabled = scopedSettings.get(INDEX_MERGE_ON_FLUSH_ENABLED);
         setMergeOnFlushPolicy(scopedSettings.get(INDEX_MERGE_ON_FLUSH_POLICY));
         defaultSearchPipeline = scopedSettings.get(DEFAULT_SEARCH_PIPELINE);
-        indexConcurrentSegmentSearchEnabled = scopedSettings.get(INDEX_CONCURRENT_SEGMENT_SEARCH_SETTING);
 
         scopedSettings.addSettingsUpdateConsumer(MergePolicyConfig.INDEX_COMPOUND_FORMAT_SETTING, mergePolicyConfig::setNoCFSRatio);
         scopedSettings.addSettingsUpdateConsumer(
@@ -913,7 +910,6 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         scopedSettings.addSettingsUpdateConsumer(INDEX_MERGE_ON_FLUSH_ENABLED, this::setMergeOnFlushEnabled);
         scopedSettings.addSettingsUpdateConsumer(INDEX_MERGE_ON_FLUSH_POLICY, this::setMergeOnFlushPolicy);
         scopedSettings.addSettingsUpdateConsumer(DEFAULT_SEARCH_PIPELINE, this::setDefaultSearchPipeline);
-        scopedSettings.addSettingsUpdateConsumer(INDEX_CONCURRENT_SEGMENT_SEARCH_SETTING, this::setIndexConcurrentSegmentSearchEnabled);
     }
 
     private void setSearchIdleAfter(TimeValue searchIdleAfter) {
@@ -1610,22 +1606,4 @@ public void setDefaultSearchPipeline(String defaultSearchPipeline) {
             );
         }
     }
-
-    public void setIndexConcurrentSegmentSearchEnabled(boolean indexConcurrentSegmentSearchEnabled) {
-        if (FeatureFlags.isEnabled(CONCURRENT_SEGMENT_SEARCH)) {
-            this.indexConcurrentSegmentSearchEnabled = indexConcurrentSegmentSearchEnabled;
-        } else {
-            throw new SettingsException(
-                "Unable to update setting: "
-                    + INDEX_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey()
-                    + ". This is an experimental feature that is currently disabled, please enable the "
-                    + CONCURRENT_SEGMENT_SEARCH
-                    + " feature flag first."
-            );
-        }
-    }
-
-    public boolean isIndexConcurrentSegmentSearchEnabled() {
-        return indexConcurrentSegmentSearchEnabled;
-    }
 }

server/src/main/java/org/opensearch/search/DefaultSearchContext.java

@@ -48,6 +48,7 @@
 import org.opensearch.common.lucene.search.Queries;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.util.BigArrays;
+import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.core.common.lease.Releasables;
 import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException;
 import org.opensearch.index.IndexService;
@@ -876,11 +877,11 @@ public Profilers getProfilers() {
      */
     @Override
     public boolean isConcurrentSegmentSearchEnabled() {
-        if (clusterService != null) {
+        if (FeatureFlags.isEnabled(FeatureFlags.CONCURRENT_SEGMENT_SEARCH) && (clusterService != null)) {
             return indexService.getIndexSettings()
                 .getSettings()
                 .getAsBoolean(
-                    "index.search.concurrent_segment_search.enabled",
+                    IndexSettings.INDEX_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey(),
                     clusterService.getClusterSettings().get(CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING)
                 );
         } else {

server/src/main/java/org/opensearch/search/SearchService.java

@@ -61,11 +61,9 @@
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Setting.Property;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.settings.SettingsException;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.util.BigArrays;
 import org.opensearch.common.util.CollectionUtils;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.common.util.concurrent.ConcurrentCollections;
 import org.opensearch.common.util.concurrent.ConcurrentMapLong;
 import org.opensearch.common.util.io.IOUtils;
@@ -159,7 +157,6 @@
 import static org.opensearch.common.unit.TimeValue.timeValueHours;
 import static org.opensearch.common.unit.TimeValue.timeValueMillis;
 import static org.opensearch.common.unit.TimeValue.timeValueMinutes;
-import static org.opensearch.common.util.FeatureFlags.CONCURRENT_SEGMENT_SEARCH;
 
 /**
  * The main search service
@@ -293,7 +290,6 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
     private volatile int maxOpenScrollContext;
 
     private volatile int maxOpenPitContext;
-    private volatile boolean clusterConcurrentSegmentSearchEnabled;
 
     private final Cancellable keepAliveReaper;
 
@@ -366,10 +362,6 @@ public SearchService(
 
         lowLevelCancellation = LOW_LEVEL_CANCELLATION_SETTING.get(settings);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(LOW_LEVEL_CANCELLATION_SETTING, this::setLowLevelCancellation);
-
-        clusterConcurrentSegmentSearchEnabled = CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING.get(settings);
-        clusterService.getClusterSettings()
-            .addSettingsUpdateConsumer(CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING, this::setClusterConcurrentSegmentSearchEnabled);
     }
 
     private void validateKeepAlives(TimeValue defaultKeepAlive, TimeValue maxKeepAlive) {
@@ -446,20 +438,6 @@ private void setLowLevelCancellation(Boolean lowLevelCancellation) {
         this.lowLevelCancellation = lowLevelCancellation;
     }
 
-    private void setClusterConcurrentSegmentSearchEnabled(boolean clusterConcurrentSegmentSearchEnabled) {
-        if (FeatureFlags.isEnabled(CONCURRENT_SEGMENT_SEARCH)) {
-            this.clusterConcurrentSegmentSearchEnabled = clusterConcurrentSegmentSearchEnabled;
-        } else {
-            throw new SettingsException(
-                "Unable to update setting: "
-                    + CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey()
-                    + ". This is an experimental feature that is currently disabled, please enable the "
-                    + CONCURRENT_SEGMENT_SEARCH
-                    + " feature flag first."
-            );
-        }
-    }
-
     @Override
     public void afterIndexRemoved(Index index, IndexSettings indexSettings, IndexRemovalReason reason) {
         // once an index is removed due to deletion or closing, we can just clean up all the pending search context information

server/src/main/java/org/opensearch/search/query/ConcurrentQueryPhaseSearcher.java

@@ -32,6 +32,7 @@
  */
 public class ConcurrentQueryPhaseSearcher extends DefaultQueryPhaseSearcher {
     private static final Logger LOGGER = LogManager.getLogger(ConcurrentQueryPhaseSearcher.class);
+    private final AggregationProcessor aggregationProcessor = new ConcurrentAggregationProcessor();
 
     /**
      * Default constructor
@@ -103,6 +104,11 @@ private static boolean searchWithCollectorManager(
         return topDocsFactory.shouldRescore();
     }
 
+    @Override
+    public AggregationProcessor aggregationProcessor(SearchContext searchContext) {
+        return aggregationProcessor;
+    }
+
     private static boolean allowConcurrentSegmentSearch(final ContextIndexSearcher searcher) {
         return (searcher.getExecutor() != null);
     }

server/src/main/java/org/opensearch/search/query/QueryPhaseSearcherWrapper.java

@@ -13,7 +13,6 @@
 import org.apache.lucene.search.Query;
 import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.search.aggregations.AggregationProcessor;
-import org.opensearch.search.aggregations.ConcurrentAggregationProcessor;
 import org.opensearch.search.internal.ContextIndexSearcher;
 import org.opensearch.search.internal.SearchContext;
 import org.apache.lucene.search.CollectorManager;
@@ -59,7 +58,7 @@ public boolean searchWith(
         boolean hasTimeout
     ) throws IOException {
         if (searchContext.isConcurrentSegmentSearchEnabled()) {
-            LOGGER.debug("Using concurrent search over segments (experimental)");
+            LOGGER.info("Using concurrent search over segments (experimental)");
             return concurrentQueryPhaseSearcher.searchWith(searchContext, searcher, query, collectors, hasFilterCollector, hasTimeout);
         } else {
             return defaultQueryPhaseSearcher.searchWith(searchContext, searcher, query, collectors, hasFilterCollector, hasTimeout);
@@ -74,8 +73,8 @@ public boolean searchWith(
     @Override
     public AggregationProcessor aggregationProcessor(SearchContext searchContext) {
         if (searchContext.isConcurrentSegmentSearchEnabled()) {
-            LOGGER.debug("Using concurrent search over segments (experimental)");
-            return new ConcurrentAggregationProcessor();
+            LOGGER.info("Using concurrent search over segments (experimental)");
+            return concurrentQueryPhaseSearcher.aggregationProcessor(searchContext);
         } else {
             return defaultQueryPhaseSearcher.aggregationProcessor(searchContext);
         }

server/src/test/java/org/opensearch/common/settings/SettingsModuleTests.java

@@ -36,6 +36,8 @@
 import org.opensearch.common.settings.Setting.Property;
 import org.hamcrest.Matchers;
 import org.opensearch.common.util.FeatureFlags;
+import org.opensearch.index.IndexSettings;
+import org.opensearch.search.SearchService;
 import org.opensearch.test.FeatureFlagSetter;
 
 import java.util.Arrays;
@@ -282,4 +284,55 @@ public void testDynamicIndexSettingsRegistration() {
             () -> module.registerDynamicSetting(Setting.floatSetting("index.custom.setting2", 1.0f, Property.IndexScope))
         );
     }
+
+    public void testConcurrentSegmentSearchClusterSettings() {
+        // Test that we throw an exception without the feature flag
+        Settings settings = Settings.builder().put(SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey(), true).build();
+        SettingsException ex = expectThrows(SettingsException.class, () -> new SettingsModule(settings));
+        assertEquals(
+            "unknown setting ["
+                + SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey()
+                + "] please check that any required plugins are installed, or check the breaking "
+                + "changes documentation for removed settings",
+            ex.getMessage()
+        );
+
+        // Test that the settings updates correctly with the feature flag
+        FeatureFlagSetter.set(FeatureFlags.CONCURRENT_SEGMENT_SEARCH);
+        boolean settingValue = randomBoolean();
+        Settings settingsWithFeatureFlag = Settings.builder()
+            .put(SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey(), settingValue)
+            .build();
+        SettingsModule settingsModule = new SettingsModule(settingsWithFeatureFlag);
+        assertEquals(settingValue, SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING.get(settingsModule.getSettings()));
+    }
+
+    public void testConcurrentSegmentSearchIndexSettings() {
+        Settings.Builder target = Settings.builder().put(Settings.EMPTY);
+        Settings.Builder update = Settings.builder();
+
+        // Test that we throw an exception without the feature flag
+        SettingsModule module = new SettingsModule(Settings.EMPTY);
+        IndexScopedSettings indexScopedSettings = module.getIndexScopedSettings();
+        expectThrows(
+            SettingsException.class,
+            () -> indexScopedSettings.updateDynamicSettings(
+                Settings.builder().put(IndexSettings.INDEX_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey(), true).build(),
+                target,
+                update,
+                "node"
+            )
+        );
+
+        // Test that the settings updates correctly with the feature flag
+        FeatureFlagSetter.set(FeatureFlags.CONCURRENT_SEGMENT_SEARCH);
+        SettingsModule moduleWithFeatureFlag = new SettingsModule(Settings.EMPTY);
+        IndexScopedSettings indexScopedSettingsWithFeatureFlag = moduleWithFeatureFlag.getIndexScopedSettings();
+        indexScopedSettingsWithFeatureFlag.updateDynamicSettings(
+            Settings.builder().put(IndexSettings.INDEX_CONCURRENT_SEGMENT_SEARCH_SETTING.getKey(), true).build(),
+            target,
+            update,
+            "node"
+        );
+    }
 }