Apply ignore_throttled also to concrete indices (#35335)

Today we only apply `ingore_throttled` to expansions from wildcards,
date math expressions and aliases. Yet, this is tricky since we might
have resolved certain expressions in pre-filter steps like security.
It's more consistent to apply this logic to all expressions including
concrete indices.

Relates to #34354

server/src/main/java/org/elasticsearch/action/support/IndicesOptions.java

@@ -248,7 +248,7 @@ public boolean ignoreAliases() {
 
     /**
      *
-     * @return whether indices that are marked as throttled should be ignored when resolving a wildcard or alias
+     * @return whether indices that are marked as throttled should be ignored
      */
     public boolean ignoreThrottled() {
         return options.contains(Option.IGNORE_THROTTLED);

server/src/main/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolver.java

@@ -147,7 +147,6 @@ Index[] concreteIndices(Context context, String... indexExpressions) {
         if (indexExpressions == null || indexExpressions.length == 0) {
             indexExpressions = new String[]{MetaData.ALL};
         }
-        Set<String> originalIndexExpression = Sets.newHashSet(indexExpressions);
         MetaData metaData = context.getState().metaData();
         IndicesOptions options = context.getOptions();
         final boolean failClosed = options.forbidClosedIndices() && options.ignoreUnavailable() == false;
@@ -197,7 +196,7 @@ Index[] concreteIndices(Context context, String... indexExpressions) {
                         " The write index may be explicitly disabled using is_write_index=false or the alias points to multiple" +
                         " indices without one being designated as a write index");
                 }
-                if (addIndex(writeIndex, context, originalIndexExpression)) {
+                if (addIndex(writeIndex, context)) {
                     concreteIndices.add(writeIndex.getIndex());
                 }
             } else {
@@ -216,12 +215,12 @@ Index[] concreteIndices(Context context, String... indexExpressions) {
                         if (failClosed) {
                             throw new IndexClosedException(index.getIndex());
                         } else {
-                            if (options.forbidClosedIndices() == false && addIndex(index, context, originalIndexExpression)) {
+                            if (options.forbidClosedIndices() == false && addIndex(index, context)) {
                                 concreteIndices.add(index.getIndex());
                             }
                         }
                     } else if (index.getState() == IndexMetaData.State.OPEN) {
-                        if (addIndex(index, context, originalIndexExpression)) {
+                        if (addIndex(index, context)) {
                             concreteIndices.add(index.getIndex());
                         }
                     } else {
@@ -239,13 +238,8 @@ Index[] concreteIndices(Context context, String... indexExpressions) {
         return concreteIndices.toArray(new Index[concreteIndices.size()]);
     }
 
-    private static boolean addIndex(IndexMetaData metaData, Context context, Set<String> originalIndices) {
-        if (context.options.ignoreThrottled()) {
-            if (originalIndices.contains(metaData.getIndex().getName()) == false) {
-                return IndexSettings.INDEX_SEARCH_THROTTLED.get(metaData.getSettings()) == false;
-            }
-        }
-        return true;
+    private static boolean addIndex(IndexMetaData metaData, Context context) {
+        return (context.options.ignoreThrottled() && IndexSettings.INDEX_SEARCH_THROTTLED.get(metaData.getSettings())) == false;
     }
 
     private static IllegalArgumentException aliasesNotSupportedException(String expression) {

server/src/test/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolverTests.java

@@ -1357,21 +1357,19 @@ public void testIgnoreThrottled() {
         {
             Index[] indices = indexNameExpressionResolver.concreteIndices(state,
                 IndicesOptions.STRICT_EXPAND_OPEN_FORBID_CLOSED_IGNORE_THROTTLED, "ind*", "test-index");
-            assertEquals(2, indices.length);
+            assertEquals(1, indices.length);
             Arrays.sort(indices, Comparator.comparing(Index::getName));
             assertEquals("index", indices[0].getName());
-            assertEquals("test-index", indices[1].getName());
         }
 
         {
             Index[] indices = indexNameExpressionResolver.concreteIndices(state,
                 new IndicesOptions(EnumSet.of(IndicesOptions.Option.ALLOW_NO_INDICES,
                     IndicesOptions.Option.IGNORE_THROTTLED),
                     EnumSet.of(IndicesOptions.WildcardStates.OPEN)), "ind*", "test-index");
-            assertEquals(2, indices.length);
+            assertEquals(1, indices.length);
             Arrays.sort(indices, Comparator.comparing(Index::getName));
             assertEquals("index", indices[0].getName());
-            assertEquals("test-index", indices[1].getName());
         }
         {
             Index[] indices = indexNameExpressionResolver.concreteIndices(state,

server/src/test/java/org/elasticsearch/search/SearchServiceTests.java

@@ -533,7 +533,8 @@ public void testSetSearchThrottled() {
         Index index = resolveIndex("throttled_threadpool_index");
         assertTrue(service.getIndicesService().indexServiceSafe(index).getIndexSettings().isSearchThrottled());
         client().prepareIndex("throttled_threadpool_index", "_doc", "1").setSource("field", "value").setRefreshPolicy(IMMEDIATE).get();
-        SearchResponse searchResponse = client().prepareSearch("throttled_threadpool_index").setSize(1).get();
+        SearchResponse searchResponse = client().prepareSearch("throttled_threadpool_index")
+            .setIndicesOptions(IndicesOptions.STRICT_EXPAND_OPEN_FORBID_CLOSED).setSize(1).get();
         assertSearchHits(searchResponse, "1");
         // we add a search action listener in a plugin above to assert that this is actually used
         client().execute(