Add a template parameter to override auto_create_index value (#61858)

Closes #20640.

This PR introduces a new parameter to v2 templates, `allow_auto_create`,
which allows templates to override the cluster setting `auto_create_index`.
Notes:

   * `AutoCreateIndex` now looks for a matching v2 template, and if its
     `allow_auto_create` setting is true, it overrides the usual logic.
   * `TransportBulkAction` previously used `AutoCreateIndex` to check
     whether missing indices should be created. We now rely on
     `AutoCreateAction`, which was already differentiating between creating
     indices and creating data streams.  I've updated `AutoCreateAction` to
     use `AutoCreateIndex`. Data streams are also influenced by
     `allow_auto_create`, in that their default auto-create behaviour can
     be disabled with this setting.
   * Most of the Java file changes are due to introducing an extra
     constructor parameter to `ComposableIndexTemplate`.
   * I've added the new setting to various x-pack templates
   * I added a YAML test to check that watches can be created even when
     `auto_create_index` is `false`.

client/rest-high-level/src/test/java/org/elasticsearch/client/IndicesClientIT.java

@@ -1579,7 +1579,7 @@ public class IndicesClientIT extends ESRestHighLevelClientTestCase {
         CompressedXContent mappings = new CompressedXContent("{\"properties\":{\"@timestamp\":{\"type\":\"date\"}}}");
         Template template = new Template(null, mappings, null);
         ComposableIndexTemplate indexTemplate = new ComposableIndexTemplate(Collections.singletonList(dataStreamName), template,
-            Collections.emptyList(), 1L, 1L, new HashMap<>(), new ComposableIndexTemplate.DataStreamTemplate());
+            Collections.emptyList(), 1L, 1L, new HashMap<>(), new ComposableIndexTemplate.DataStreamTemplate(), null);
         PutComposableIndexTemplateRequest putComposableIndexTemplateRequest =
             new PutComposableIndexTemplateRequest().name("ds-template").create(true).indexTemplate(indexTemplate);
         AcknowledgedResponse response = execute(putComposableIndexTemplateRequest,
@@ -1658,7 +1658,7 @@ public class IndicesClientIT extends ESRestHighLevelClientTestCase {
         Template template = new Template(settings, mappings, Map.of("alias", alias));
         List<String> pattern = List.of("pattern");
         ComposableIndexTemplate indexTemplate =
-            new ComposableIndexTemplate(pattern, template, Collections.emptyList(), 1L, 1L, new HashMap<>(), null);
+            new ComposableIndexTemplate(pattern, template, Collections.emptyList(), 1L, 1L, new HashMap<>(), null, null);
         PutComposableIndexTemplateRequest putComposableIndexTemplateRequest =
             new PutComposableIndexTemplateRequest().name(templateName).create(true).indexTemplate(indexTemplate);
 
@@ -1705,7 +1705,7 @@ public class IndicesClientIT extends ESRestHighLevelClientTestCase {
         Template template = new Template(settings, mappings, Map.of("alias", alias));
         List<String> pattern = List.of("pattern");
         ComposableIndexTemplate indexTemplate =
-            new ComposableIndexTemplate(pattern, template, Collections.emptyList(), 1L, 1L, new HashMap<>(), null);
+            new ComposableIndexTemplate(pattern, template, Collections.emptyList(), 1L, 1L, new HashMap<>(), null, null);
         PutComposableIndexTemplateRequest putComposableIndexTemplateRequest =
             new PutComposableIndexTemplateRequest().name(templateName).create(true).indexTemplate(indexTemplate);
 
@@ -1716,7 +1716,7 @@ public class IndicesClientIT extends ESRestHighLevelClientTestCase {
         SimulateIndexTemplateRequest simulateIndexTemplateRequest = new SimulateIndexTemplateRequest("pattern");
         AliasMetadata simulationAlias = AliasMetadata.builder("simulation-alias").writeIndex(true).build();
         ComposableIndexTemplate simulationTemplate = new ComposableIndexTemplate(pattern, new Template(null, null,
-            Map.of("simulation-alias", simulationAlias)), Collections.emptyList(), 2L, 1L, new HashMap<>(), null);
+            Map.of("simulation-alias", simulationAlias)), Collections.emptyList(), 2L, 1L, new HashMap<>(), null, null);
         PutComposableIndexTemplateRequest newIndexTemplateReq =
             new PutComposableIndexTemplateRequest().name("used-for-simulation").create(true).indexTemplate(indexTemplate);
         newIndexTemplateReq.indexTemplate(simulationTemplate);

client/rest-high-level/src/test/java/org/elasticsearch/client/indices/GetComposableIndexTemplatesResponseTests.java

@@ -88,6 +88,6 @@ public class GetComposableIndexTemplatesResponseTests extends ESTestCase {
         if (randomBoolean()) {
             dataStreamTemplate = new ComposableIndexTemplate.DataStreamTemplate();
         }
-        return new ComposableIndexTemplate(patterns, randomTemplate(), composedOf, priority, version, meta, dataStreamTemplate);
+        return new ComposableIndexTemplate(patterns, randomTemplate(), composedOf, priority, version, meta, dataStreamTemplate, null);
     }
 }

docs/reference/indices/put-component-template.asciidoc

@@ -4,10 +4,10 @@
 <titleabbrev>Put component template</titleabbrev>
 ++++
 
-Creates or updates a component template. 
-Component templates are building blocks for constructing <<indices-templates,index templates>>.  
-that specify index <<mapping,mappings>>, <<index-modules-settings,settings>>, 
-and <<indices-aliases,aliases>>. 
+Creates or updates a component template.
+Component templates are building blocks for constructing <<indices-templates,index templates>>.
+that specify index <<mapping,mappings>>, <<index-modules-settings,settings>>,
+and <<indices-aliases,aliases>>.
 
 [source,console]
 --------------------------------------------------
@@ -55,10 +55,10 @@ DELETE _component_template/template_*
 [[put-component-template-api-desc]]
 ==== {api-description-title}
 
-An index template can be composed of multiple component templates. 
+An index template can be composed of multiple component templates.
 To use a component template, specify it in an index template's `composed_of` list.
-Component templates are only applied to new data streams and indices 
-as part of a matching index template. 
+Component templates are only applied to new data streams and indices
+as part of a matching index template.
 
 Settings and mappings specified directly in the index template or the <<indices-create-index, create index>>
 request override any settings or mappings specified in a component template.
@@ -112,6 +112,16 @@ include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=settings]
 Version number used to manage component templates externally.
 This number is not automatically generated or incremented by {es}.
 
+`allow_auto_create`::
+(Optional, boolean)
+This setting overrides the value of the
+<<index-creation,`action.auto_create_index`>> cluster setting. If set to
+`true` in a template, then indices can be automatically created using that
+template even if auto-creation of indices is disabled via
+`actions.auto_create_index`. If set to `false`, then indices or data streams matching the
+template must always be explicitly created, and may never be automatically
+created.
+
 `_meta`::
 (Optional, object)
 Optional user metadata about the component template. May have any contents.
@@ -157,7 +167,7 @@ To be applied, a component template must be included in an index template's `com
 [[component-templates-version]]
 ===== Component template versioning
 
-You can use the `version` parameter to add a version number to a component template. 
+You can use the `version` parameter to add a version number to a component template.
 External systems can use these version numbers to simplify template management.
 
 The `version` parameter is optional and not automatically generated or used by {es}.
@@ -182,7 +192,7 @@ To check the `version`, you can use the <<getting-component-templates,get compon
 [[component-templates-metadata]]
 ===== Component template metadata
 
-You can use the `_meta` parameter to add arbitrary metadata to a component template. 
+You can use the `_meta` parameter to add arbitrary metadata to a component template.
 This user-defined object is stored in the cluster state,
 so keeping it short is preferrable.
 

qa/smoke-test-http/src/test/java/org/elasticsearch/http/AutoCreateIndexIT.java

@@ -0,0 +1,126 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.http;
+
+import org.elasticsearch.action.support.AutoCreateIndex;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.Streams;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.test.rest.ESRestTestCase;
+
+import java.io.IOException;
+import java.io.InputStreamReader;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.hamcrest.Matchers.containsString;
+
+public class AutoCreateIndexIT extends ESRestTestCase {
+
+    /**
+     * Check that setting {@link AutoCreateIndex#AUTO_CREATE_INDEX_SETTING} to <code>false</code>
+     * disable the automatic creation on indices.
+     */
+    public void testCannotAutoCreateIndexWhenDisabled() throws IOException {
+        configureAutoCreateIndex(false);
+
+        // Attempt to add a document to a non-existing index. Auto-creating the index should fail owing to the setting above.
+        final Request indexDocumentRequest = new Request("POST", "recipe_kr/_doc/123456");
+        indexDocumentRequest.setJsonEntity("{ \"name\": \"Kimchi\" }");
+        final ResponseException responseException = expectThrows(ResponseException.class, this::indexDocument);
+
+        assertThat(
+            Streams.copyToString(new InputStreamReader(responseException.getResponse().getEntity().getContent(), UTF_8)),
+            containsString("no such index [recipe_kr] and [action.auto_create_index] is [false]")
+        );
+    }
+
+    /**
+     * Check that automatically creating an index is allowed, even when {@link AutoCreateIndex#AUTO_CREATE_INDEX_SETTING}
+     * is <code>false</code>, when the index name matches a template and that template has <code>allow_auto_create</code>
+     * set to <code>true</code>.
+     */
+    public void testCanAutoCreateIndexWhenAllowedByTemplate() throws IOException {
+        configureAutoCreateIndex(false);
+
+        createTemplateWithAllowAutoCreate(true);
+
+        // Attempt to add a document to a non-existing index. Auto-creating the index should succeed because the index name
+        // matches the template pattern
+        assertOK(this.indexDocument());
+    }
+
+    /**
+     * Check that automatically creating an index is disallowed when the index name matches a template and that template has
+     * <code>allow_auto_create</code> explicitly to <code>false</code>, even when {@link AutoCreateIndex#AUTO_CREATE_INDEX_SETTING}
+     * is set to <code>true</code>.
+     */
+    public void testCannotAutoCreateIndexWhenDisallowedByTemplate() throws IOException {
+        configureAutoCreateIndex(true);
+
+        createTemplateWithAllowAutoCreate(false);
+
+        // Attempt to add a document to a non-existing index. Auto-creating the index should succeed because the index name
+        // matches the template pattern
+        final ResponseException responseException = expectThrows(ResponseException.class, this::indexDocument);
+
+        assertThat(
+            Streams.copyToString(new InputStreamReader(responseException.getResponse().getEntity().getContent(), UTF_8)),
+            containsString("no such index [composable template [recipe*] forbids index auto creation]")
+        );
+    }
+
+
+    private void configureAutoCreateIndex(boolean value) throws IOException {
+        XContentBuilder builder = JsonXContent.contentBuilder()
+            .startObject()
+            .startObject("transient")
+            .field(AutoCreateIndex.AUTO_CREATE_INDEX_SETTING.getKey(), value)
+            .endObject()
+            .endObject();
+
+        final Request settingsRequest = new Request("PUT", "_cluster/settings");
+        settingsRequest.setJsonEntity(Strings.toString(builder));
+        final Response settingsResponse = client().performRequest(settingsRequest);
+        assertOK(settingsResponse);
+    }
+
+    private void createTemplateWithAllowAutoCreate(Boolean allowAutoCreate) throws IOException {
+        XContentBuilder builder = JsonXContent.contentBuilder()
+            .startObject()
+            .array("index_patterns", "recipe*")
+            .field("allow_auto_create", allowAutoCreate)
+            .endObject();
+
+        final Request createTemplateRequest = new Request("PUT", "_index_template/recipe_template");
+        createTemplateRequest.setJsonEntity(Strings.toString(builder));
+        final Response createTemplateResponse = client().performRequest(createTemplateRequest);
+        assertOK(createTemplateResponse);
+    }
+
+    private Response indexDocument() throws IOException {
+        final Request indexDocumentRequest = new Request("POST", "recipe_kr/_doc/123456");
+        indexDocumentRequest.setJsonEntity("{ \"name\": \"Kimchi\" }");
+        return client().performRequest(indexDocumentRequest);
+    }
+}

server/src/internalClusterTest/java/org/elasticsearch/action/bulk/BulkProcessorClusterSettingsIT.java

@@ -49,7 +49,8 @@ public class BulkProcessorClusterSettingsIT extends ESIntegTestCase {
         assertTrue("Missing index should have been flagged", responses[1].isFailed());
         assertThat(
             responses[1].getFailureMessage(),
-            equalTo("[wontwork] org.elasticsearch.index.IndexNotFoundException: no such index [wontwork]"));
+            equalTo("[wontwork] org.elasticsearch.index.IndexNotFoundException: no such index [wontwork]"
+                + " and [action.auto_create_index] is [false]"));
         assertFalse("Operation on existing index should succeed", responses[2].isFailed());
     }
 }

server/src/internalClusterTest/java/org/elasticsearch/indices/template/ComposableTemplateIT.java

@@ -102,7 +102,7 @@ public class ComposableTemplateIT extends ESIntegTestCase {
         Exception expectedException = expectThrows(Exception.class, () -> ComposableIndexTemplate.parse(parser));
 
         ComposableIndexTemplate template = new ComposableIndexTemplate(List.of("logs-*-*"), null, null, null, null,
-            null, new ComposableIndexTemplate.DataStreamTemplate());
+            null, new ComposableIndexTemplate.DataStreamTemplate(), null);
         Exception e = expectThrows(IllegalArgumentException.class, () -> client().execute(PutComposableIndexTemplateAction.INSTANCE,
             new PutComposableIndexTemplateAction.Request("my-it").indexTemplate(template)).actionGet());
         Exception actualException = (Exception) e.getCause();

server/src/main/java/org/elasticsearch/action/admin/indices/create/AutoCreateAction.java

@@ -23,15 +23,15 @@ import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.action.support.ActiveShardsObserver;
+import org.elasticsearch.action.support.AutoCreateIndex;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ack.ClusterStateUpdateResponse;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
-import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
-import org.elasticsearch.cluster.metadata.ComposableIndexTemplate.DataStreamTemplate;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.MetadataCreateDataStreamService;
 import org.elasticsearch.cluster.metadata.MetadataCreateDataStreamService.CreateDataStreamClusterStateUpdateRequest;
@@ -40,6 +40,7 @@ import org.elasticsearch.cluster.metadata.MetadataIndexTemplateService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -63,17 +64,20 @@ public final class AutoCreateAction extends ActionType<CreateIndexResponse> {
         private final ActiveShardsObserver activeShardsObserver;
         private final MetadataCreateIndexService createIndexService;
         private final MetadataCreateDataStreamService metadataCreateDataStreamService;
+        private final AutoCreateIndex autoCreateIndex;
 
         @Inject
         public TransportAction(TransportService transportService, ClusterService clusterService, ThreadPool threadPool,
                                ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
                                MetadataCreateIndexService createIndexService,
-                               MetadataCreateDataStreamService metadataCreateDataStreamService) {
+                               MetadataCreateDataStreamService metadataCreateDataStreamService,
+                               AutoCreateIndex autoCreateIndex) {
             super(NAME, transportService, clusterService, threadPool, actionFilters, CreateIndexRequest::new, indexNameExpressionResolver,
                     CreateIndexResponse::new, ThreadPool.Names.SAME);
             this.activeShardsObserver = new ActiveShardsObserver(clusterService, threadPool);
             this.createIndexService = createIndexService;
             this.metadataCreateDataStreamService = metadataCreateDataStreamService;
+            this.autoCreateIndex = autoCreateIndex;
         }
 
         @Override
@@ -112,16 +116,36 @@ public final class AutoCreateAction extends ActionType<CreateIndexResponse> {
 
                 @Override
                 public ClusterState execute(ClusterState currentState) throws Exception {
-                    DataStreamTemplate dataStreamTemplate = resolveAutoCreateDataStream(request, currentState.metadata());
-                    if (dataStreamTemplate != null) {
+                    final ComposableIndexTemplate template = resolveTemplate(request, currentState.metadata());
+
+                    if (template != null && template.getDataStreamTemplate() != null) {
+                        // This expression only evaluates to true when the argument is non-null and false
+                        if (Boolean.FALSE.equals(template.getAllowAutoCreate())) {
+                            throw new IndexNotFoundException(
+                                "composable template " + template.indexPatterns() + " forbids index auto creation"
+                            );
+                        }
+
                         CreateDataStreamClusterStateUpdateRequest createRequest = new CreateDataStreamClusterStateUpdateRequest(
-                            request.index(), request.masterNodeTimeout(), request.timeout());
-                        ClusterState clusterState =  metadataCreateDataStreamService.createDataStream(createRequest, currentState);
+                            request.index(),
+                            request.masterNodeTimeout(),
+                            request.timeout()
+                        );
+                        ClusterState clusterState = metadataCreateDataStreamService.createDataStream(createRequest, currentState);
                         indexNameRef.set(clusterState.metadata().dataStreams().get(request.index()).getIndices().get(0).getName());
                         return clusterState;
                     } else {
                         String indexName = indexNameExpressionResolver.resolveDateMathExpression(request.index());
                         indexNameRef.set(indexName);
+
+                        // This will throw an exception if the index does not exist and creating it is prohibited
+                        final boolean shouldAutoCreate = autoCreateIndex.shouldAutoCreate(indexName, currentState);
+
+                        if (shouldAutoCreate == false) {
+                            // The index already exists.
+                            return currentState;
+                        }
+
                         CreateIndexClusterStateUpdateRequest updateRequest =
                             new CreateIndexClusterStateUpdateRequest(request.cause(), indexName, request.index())
                                 .ackTimeout(request.timeout()).masterNodeTimeout(request.masterNodeTimeout());
@@ -137,16 +161,8 @@ public final class AutoCreateAction extends ActionType<CreateIndexResponse> {
         }
     }
 
-    static DataStreamTemplate resolveAutoCreateDataStream(CreateIndexRequest request, Metadata metadata) {
+    static ComposableIndexTemplate resolveTemplate(CreateIndexRequest request, Metadata metadata) {
         String v2Template = MetadataIndexTemplateService.findV2Template(metadata, request.index(), false);
-        if (v2Template != null) {
-            ComposableIndexTemplate composableIndexTemplate = metadata.templatesV2().get(v2Template);
-            if (composableIndexTemplate.getDataStreamTemplate() != null) {
-                return composableIndexTemplate.getDataStreamTemplate();
-            }
-        }
-
-        return null;
+        return v2Template != null ? metadata.templatesV2().get(v2Template) : null;
     }
-
 }

server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java

@@ -38,7 +38,6 @@ import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.ingest.IngestActionForwarder;
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.AutoCreateIndex;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.update.TransportUpdateAction;
 import org.elasticsearch.action.update.UpdateRequest;
@@ -90,7 +89,6 @@ import java.util.concurrent.atomic.AtomicIntegerArray;
 import java.util.function.LongSupplier;
 import java.util.stream.Collectors;
 
-import static java.util.Collections.emptyMap;
 import static org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.EXCLUDED_DATA_STREAMS_KEY;
 import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_PRIMARY_TERM;
 import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_SEQ_NO;
@@ -104,7 +102,6 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
     private static final Logger logger = LogManager.getLogger(TransportBulkAction.class);
 
     private final ThreadPool threadPool;
-    private final AutoCreateIndex autoCreateIndex;
     private final ClusterService clusterService;
     private final IngestService ingestService;
     private final LongSupplier relativeTimeProvider;
@@ -119,22 +116,21 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
     public TransportBulkAction(ThreadPool threadPool, TransportService transportService,
                                ClusterService clusterService, IngestService ingestService,
                                NodeClient client, ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
-                               AutoCreateIndex autoCreateIndex, IndexingPressure indexingPressure, SystemIndices systemIndices) {
+                               IndexingPressure indexingPressure, SystemIndices systemIndices) {
         this(threadPool, transportService, clusterService, ingestService, client, actionFilters,
-            indexNameExpressionResolver, autoCreateIndex, indexingPressure, systemIndices, System::nanoTime);
+            indexNameExpressionResolver, indexingPressure, systemIndices, System::nanoTime);
     }
 
     public TransportBulkAction(ThreadPool threadPool, TransportService transportService,
                                ClusterService clusterService, IngestService ingestService,
                                NodeClient client, ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
-                               AutoCreateIndex autoCreateIndex, IndexingPressure indexingPressure, SystemIndices systemIndices,
+                               IndexingPressure indexingPressure, SystemIndices systemIndices,
                                LongSupplier relativeTimeProvider) {
         super(BulkAction.NAME, transportService, actionFilters, BulkRequest::new, ThreadPool.Names.SAME);
         Objects.requireNonNull(relativeTimeProvider);
         this.threadPool = threadPool;
         this.clusterService = clusterService;
         this.ingestService = ingestService;
-        this.autoCreateIndex = autoCreateIndex;
         this.relativeTimeProvider = relativeTimeProvider;
         this.ingestForwarder = new IngestActionForwarder(transportService);
         this.client = client;
@@ -224,76 +220,69 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
             return;
         }
 
-        final boolean includesSystem = includesSystem(bulkRequest, clusterService.state().metadata().getIndicesLookup(), systemIndices);
-
-        if (includesSystem || needToCheck()) {
-            // Attempt to create all the indices that we're going to need during the bulk before we start.
-            // Step 1: collect all the indices in the request
-            final Map<String, Boolean> indices = bulkRequest.requests.stream()
-                // delete requests should not attempt to create the index (if the index does not
-                // exists), unless an external versioning is used
-                .filter(request -> request.opType() != DocWriteRequest.OpType.DELETE
-                    || request.versionType() == VersionType.EXTERNAL
-                    || request.versionType() == VersionType.EXTERNAL_GTE)
-                .collect(Collectors.toMap(DocWriteRequest::index, DocWriteRequest::isRequireAlias, (v1, v2) -> v1 || v2));
-            /* Step 2: filter that to indices that don't exist and we can create. At the same time build a map of indices we can't create
-             * that we'll use when we try to run the requests. */
-            final Map<String, IndexNotFoundException> indicesThatCannotBeCreated = new HashMap<>();
-            Set<String> autoCreateIndices = new HashSet<>();
-            ClusterState state = clusterService.state();
-            for (Map.Entry<String, Boolean> indexAndFlag : indices.entrySet()) {
-                boolean shouldAutoCreate;
-                final String index = indexAndFlag.getKey();
-                try {
-                    shouldAutoCreate = shouldAutoCreate(index, state);
-                } catch (IndexNotFoundException e) {
-                    shouldAutoCreate = false;
-                    indicesThatCannotBeCreated.put(index, e);
-                }
-                // We should only auto create if we are not requiring it to be an alias
-                if (shouldAutoCreate && (indexAndFlag.getValue() == false)) {
-                    autoCreateIndices.add(index);
-                }
+        // Attempt to create all the indices that we're going to need during the bulk before we start.
+        // Step 1: collect all the indices in the request
+        final Map<String, Boolean> indices = bulkRequest.requests.stream()
+            // delete requests should not attempt to create the index (if the index does not
+            // exists), unless an external versioning is used
+            .filter(request -> request.opType() != DocWriteRequest.OpType.DELETE
+                || request.versionType() == VersionType.EXTERNAL
+                || request.versionType() == VersionType.EXTERNAL_GTE)
+            .collect(Collectors.toMap(DocWriteRequest::index, DocWriteRequest::isRequireAlias, (v1, v2) -> v1 || v2));
+
+        // Step 2: filter the list of indices to find those that don't currently exist.
+        final Map<String, IndexNotFoundException> indicesThatCannotBeCreated = new HashMap<>();
+        Set<String> autoCreateIndices = new HashSet<>();
+        ClusterState state = clusterService.state();
+        for (Map.Entry<String, Boolean> indexAndFlag : indices.entrySet()) {
+            final String index = indexAndFlag.getKey();
+            boolean shouldAutoCreate = indexNameExpressionResolver.hasIndexAbstraction(index, state) == false;
+            // We should only auto create if we are not requiring it to be an alias
+            if (shouldAutoCreate && (indexAndFlag.getValue() == false)) {
+                autoCreateIndices.add(index);
             }
-            // Step 3: create all the indices that are missing, if there are any missing. start the bulk after all the creates come back.
-            if (autoCreateIndices.isEmpty()) {
-                executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated);
-            } else {
-                final AtomicInteger counter = new AtomicInteger(autoCreateIndices.size());
-                for (String index : autoCreateIndices) {
-                    createIndex(index, bulkRequest.timeout(), minNodeVersion, new ActionListener<>() {
-                        @Override
-                        public void onResponse(CreateIndexResponse result) {
-                            if (counter.decrementAndGet() == 0) {
-                                threadPool.executor(executorName).execute(
-                                    () -> executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated));
-                            }
+        }
+
+        // Step 3: create all the indices that are missing, if there are any missing. start the bulk after all the creates come back.
+        if (autoCreateIndices.isEmpty()) {
+            executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated);
+        } else {
+            final AtomicInteger counter = new AtomicInteger(autoCreateIndices.size());
+            for (String index : autoCreateIndices) {
+                createIndex(index, bulkRequest.timeout(), minNodeVersion, new ActionListener<>() {
+                    @Override
+                    public void onResponse(CreateIndexResponse result) {
+                        if (counter.decrementAndGet() == 0) {
+                            threadPool.executor(executorName).execute(
+                                () -> executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated));
                         }
+                    }
 
-                        @Override
-                        public void onFailure(Exception e) {
-                            if (!(ExceptionsHelper.unwrapCause(e) instanceof ResourceAlreadyExistsException)) {
-                                // fail all requests involving this index, if create didn't work
-                                for (int i = 0; i < bulkRequest.requests.size(); i++) {
-                                    DocWriteRequest<?> request = bulkRequest.requests.get(i);
-                                    if (request != null && setResponseFailureIfIndexMatches(responses, i, request, index, e)) {
-                                        bulkRequest.requests.set(i, null);
-                                    }
+                    @Override
+                    public void onFailure(Exception e) {
+                        final Throwable cause = ExceptionsHelper.unwrapCause(e);
+                        if (cause instanceof IndexNotFoundException) {
+                            indicesThatCannotBeCreated.put(index, (IndexNotFoundException) e);
+                        }
+                        else if ((cause instanceof ResourceAlreadyExistsException) == false) {
+                            // fail all requests involving this index, if create didn't work
+                            for (int i = 0; i < bulkRequest.requests.size(); i++) {
+                                DocWriteRequest<?> request = bulkRequest.requests.get(i);
+                                if (request != null && setResponseFailureIfIndexMatches(responses, i, request, index, e)) {
+                                    bulkRequest.requests.set(i, null);
                                 }
                             }
-                            if (counter.decrementAndGet() == 0) {
-                                threadPool.executor(executorName).execute(() -> executeBulk(task, bulkRequest, startTime,
-                                    ActionListener.wrap(listener::onResponse, inner -> {
+                        }
+                        if (counter.decrementAndGet() == 0) {
+                            threadPool.executor(executorName).execute(() -> executeBulk(task, bulkRequest, startTime,
+                                ActionListener.wrap(listener::onResponse, inner -> {
                                     inner.addSuppressed(e);
                                     listener.onFailure(inner);
                                 }), responses, indicesThatCannotBeCreated));
-                            }
                         }
-                    });
-                }
+                    }
+                });
             }
-        } else {
-            executeBulk(task, bulkRequest, startTime, listener, responses, emptyMap());
         }
     }
 
@@ -351,10 +340,6 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
         return request.getIndices().stream().allMatch(indexName -> isSystemIndex(indicesLookup, systemIndices, indexName));
     }
 
-    boolean includesSystem(BulkRequest request, SortedMap<String, IndexAbstraction> indicesLookup, SystemIndices systemIndices) {
-        return request.getIndices().stream().anyMatch(indexName -> isSystemIndex(indicesLookup, systemIndices, indexName));
-    }
-
     private boolean isSystemIndex(SortedMap<String, IndexAbstraction> indicesLookup, SystemIndices systemIndices, String indexName) {
         final IndexAbstraction abstraction = indicesLookup.get(indexName);
         if (abstraction != null) {
@@ -364,14 +349,6 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
         }
     }
 
-    boolean needToCheck() {
-        return autoCreateIndex.needToCheck();
-    }
-
-    boolean shouldAutoCreate(String index, ClusterState state) {
-        return autoCreateIndex.shouldAutoCreate(index, state);
-    }
-
     void createIndex(String index,
                      TimeValue timeout,
                      Version minNodeVersion,

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

@@ -20,7 +20,10 @@
 package org.elasticsearch.action.support;
 
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.metadata.MetadataIndexTemplateService;
 import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.Tuple;
@@ -40,7 +43,6 @@ import java.util.List;
  * a write operation is about to happen in a non existing index.
  */
 public final class AutoCreateIndex {
-
     public static final Setting<AutoCreate> AUTO_CREATE_INDEX_SETTING =
         new Setting<>("action.auto_create_index", "true", AutoCreate::new, Property.NodeScope, Setting.Property.Dynamic);
 
@@ -58,13 +60,6 @@ public final class AutoCreateIndex {
         clusterSettings.addSettingsUpdateConsumer(AUTO_CREATE_INDEX_SETTING, this::setAutoCreate);
     }
 
-    /**
-     * Do we really need to check if an index should be auto created?
-     */
-    public boolean needToCheck() {
-        return this.autoCreate.autoCreateIndex;
-    }
-
     /**
      * Should the index be auto created?
      * @throws IndexNotFoundException if the index doesn't exist and shouldn't be auto created
@@ -79,6 +74,17 @@ public final class AutoCreateIndex {
             return true;
         }
 
+        // Templates can override the AUTO_CREATE_INDEX_SETTING setting
+        final ComposableIndexTemplate template = findTemplate(index, state.metadata());
+        if (template != null && template.getAllowAutoCreate() != null) {
+            if (template.getAllowAutoCreate()) {
+                return true;
+            } else {
+                // An explicit false value overrides AUTO_CREATE_INDEX_SETTING
+                throw new IndexNotFoundException("composable template " + template.indexPatterns() + " forbids index auto creation");
+            }
+        }
+
         // One volatile read, so that all checks are done against the same instance:
         final AutoCreate autoCreate = this.autoCreate;
         if (autoCreate.autoCreateIndex == false) {
@@ -112,6 +118,11 @@ public final class AutoCreateIndex {
         this.autoCreate = autoCreate;
     }
 
+    private ComposableIndexTemplate findTemplate(String indexName, Metadata metadata) {
+        final String templateName = MetadataIndexTemplateService.findV2Template(metadata, indexName, false);
+        return metadata.templatesV2().get(templateName);
+    }
+
     static class AutoCreate {
         private final boolean autoCreateIndex;
         private final List<Tuple<String, Boolean>> expressions;

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

@@ -55,6 +55,9 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
     private static final ParseField VERSION = new ParseField("version");
     private static final ParseField METADATA = new ParseField("_meta");
     private static final ParseField DATA_STREAM = new ParseField("data_stream");
+    private static final ParseField ALLOW_AUTO_CREATE = new ParseField("allow_auto_create");
+
+    private static final Version ALLOW_AUTO_CREATE_VERSION = Version.V_8_0_0;
 
     @SuppressWarnings("unchecked")
     public static final ConstructingObjectParser<ComposableIndexTemplate, Void> PARSER = new ConstructingObjectParser<>("index_template",
@@ -65,7 +68,8 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
             (Long) a[3],
             (Long) a[4],
             (Map<String, Object>) a[5],
-            (DataStreamTemplate) a[6]));
+            (DataStreamTemplate) a[6],
+            (Boolean) a[7]));
 
     static {
         PARSER.declareStringArray(ConstructingObjectParser.constructorArg(), INDEX_PATTERNS);
@@ -75,6 +79,7 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
         PARSER.declareLong(ConstructingObjectParser.optionalConstructorArg(), VERSION);
         PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> p.map(), METADATA);
         PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), DataStreamTemplate.PARSER, DATA_STREAM);
+        PARSER.declareBoolean(ConstructingObjectParser.optionalConstructorArg(), ALLOW_AUTO_CREATE);
     }
 
     private final List<String> indexPatterns;
@@ -90,6 +95,8 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
     private final Map<String, Object> metadata;
     @Nullable
     private final DataStreamTemplate dataStreamTemplate;
+    @Nullable
+    private final Boolean allowAutoCreate;
 
     static Diff<ComposableIndexTemplate> readITV2DiffFrom(StreamInput in) throws IOException {
         return AbstractDiffable.readDiffFrom(ComposableIndexTemplate::new, in);
@@ -101,12 +108,12 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
 
     public ComposableIndexTemplate(List<String> indexPatterns, @Nullable Template template, @Nullable List<String> componentTemplates,
                                    @Nullable Long priority, @Nullable Long version, @Nullable Map<String, Object> metadata) {
-        this(indexPatterns, template, componentTemplates, priority, version, metadata, null);
+        this(indexPatterns, template, componentTemplates, priority, version, metadata, null, null);
     }
 
     public ComposableIndexTemplate(List<String> indexPatterns, @Nullable Template template, @Nullable List<String> componentTemplates,
                                    @Nullable Long priority, @Nullable Long version, @Nullable Map<String, Object> metadata,
-                                   @Nullable DataStreamTemplate dataStreamTemplate) {
+                                   @Nullable DataStreamTemplate dataStreamTemplate, @Nullable Boolean allowAutoCreate) {
         this.indexPatterns = indexPatterns;
         this.template = template;
         this.componentTemplates = componentTemplates;
@@ -114,6 +121,7 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
         this.version = version;
         this.metadata = metadata;
         this.dataStreamTemplate = dataStreamTemplate;
+        this.allowAutoCreate = allowAutoCreate;
     }
 
     public ComposableIndexTemplate(StreamInput in) throws IOException {
@@ -132,6 +140,11 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
         } else {
             this.dataStreamTemplate = null;
         }
+        if (in.getVersion().onOrAfter(ALLOW_AUTO_CREATE_VERSION)) {
+            this.allowAutoCreate = in.readOptionalBoolean();
+        } else {
+            this.allowAutoCreate = null;
+        }
     }
 
     public List<String> indexPatterns() {
@@ -173,6 +186,11 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
         return dataStreamTemplate;
     }
 
+    @Nullable
+    public Boolean getAllowAutoCreate() {
+        return this.allowAutoCreate;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeStringCollection(this.indexPatterns);
@@ -189,6 +207,9 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
         if (out.getVersion().onOrAfter(Version.V_7_9_0)) {
             out.writeOptionalWriteable(dataStreamTemplate);
         }
+        if (out.getVersion().onOrAfter(ALLOW_AUTO_CREATE_VERSION)) {
+            out.writeOptionalBoolean(allowAutoCreate);
+        }
     }
 
     @Override
@@ -213,6 +234,9 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
         if (this.dataStreamTemplate != null) {
             builder.field(DATA_STREAM.getPreferredName(), dataStreamTemplate);
         }
+        if (this.allowAutoCreate != null) {
+            builder.field(ALLOW_AUTO_CREATE.getPreferredName(), allowAutoCreate);
+        }
         builder.endObject();
         return builder;
     }
@@ -220,7 +244,7 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
     @Override
     public int hashCode() {
         return Objects.hash(this.indexPatterns, this.template, this.componentTemplates, this.priority, this.version,
-            this.metadata, this.dataStreamTemplate);
+            this.metadata, this.dataStreamTemplate, this.allowAutoCreate);
     }
 
     @Override
@@ -238,7 +262,8 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
             Objects.equals(this.priority, other.priority) &&
             Objects.equals(this.version, other.version) &&
             Objects.equals(this.metadata, other.metadata) &&
-            Objects.equals(this.dataStreamTemplate, other.dataStreamTemplate);
+            Objects.equals(this.dataStreamTemplate, other.dataStreamTemplate) &&
+            Objects.equals(this.allowAutoCreate, other.allowAutoCreate);
     }
 
     @Override

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

@@ -501,7 +501,8 @@ public class MetadataIndexTemplateService {
             final Template finalTemplate = new Template(finalSettings,
                 stringMappings == null ? null : new CompressedXContent(stringMappings), innerTemplate.aliases());
             finalIndexTemplate = new ComposableIndexTemplate(template.indexPatterns(), finalTemplate, template.composedOf(),
-                template.priority(), template.version(), template.metadata(), template.getDataStreamTemplate());
+                template.priority(), template.version(), template.metadata(), template.getDataStreamTemplate(),
+                template.getAllowAutoCreate());
         }
 
         if (finalIndexTemplate.equals(existing)) {

server/src/test/java/org/elasticsearch/action/admin/indices/create/AutoCreateActionTests.java

@@ -32,35 +32,38 @@ import static org.hamcrest.Matchers.nullValue;
 
 public class AutoCreateActionTests extends ESTestCase {
 
-    public void testResolveAutoCreateDataStreams() {
+    public void testResolveTemplates() {
         Metadata metadata;
         {
             Metadata.Builder mdBuilder = new Metadata.Builder();
             DataStreamTemplate dataStreamTemplate = new DataStreamTemplate();
-            mdBuilder.put("1", new ComposableIndexTemplate(List.of("legacy-logs-*"), null, null, 10L, null, null, null));
-            mdBuilder.put("2", new ComposableIndexTemplate(List.of("logs-*"), null, null, 20L, null, null, dataStreamTemplate));
-            mdBuilder.put("3", new ComposableIndexTemplate(List.of("logs-foobar"), null, null, 30L, null, null, dataStreamTemplate));
+            mdBuilder.put("1", new ComposableIndexTemplate(List.of("legacy-logs-*"), null, null, 10L, null, null, null, null));
+            mdBuilder.put("2", new ComposableIndexTemplate(List.of("logs-*"), null, null, 20L, null, null, dataStreamTemplate, null));
+            mdBuilder.put("3", new ComposableIndexTemplate(List.of("logs-foobar"), null, null, 30L, null, null, dataStreamTemplate, null));
             metadata = mdBuilder.build();
         }
 
         CreateIndexRequest request = new CreateIndexRequest("logs-foobar");
-        DataStreamTemplate result  = AutoCreateAction.resolveAutoCreateDataStream(request, metadata);
+        ComposableIndexTemplate result  = AutoCreateAction.resolveTemplate(request, metadata);
         assertThat(result, notNullValue());
-        assertThat(result.getTimestampField(), equalTo("@timestamp"));
+        assertThat(result.getDataStreamTemplate(), notNullValue());
+        assertThat(result.getDataStreamTemplate().getTimestampField(), equalTo("@timestamp"));
 
         request = new CreateIndexRequest("logs-barbaz");
-        result  = AutoCreateAction.resolveAutoCreateDataStream(request, metadata);
+        result  = AutoCreateAction.resolveTemplate(request, metadata);
         assertThat(result, notNullValue());
-        assertThat(result.getTimestampField(), equalTo("@timestamp"));
+        assertThat(result.getDataStreamTemplate(), notNullValue());
+        assertThat(result.getDataStreamTemplate().getTimestampField(), equalTo("@timestamp"));
 
         // An index that matches with a template without a data steam definition
         request = new CreateIndexRequest("legacy-logs-foobaz");
-        result = AutoCreateAction.resolveAutoCreateDataStream(request, metadata);
-        assertThat(result, nullValue());
+        result = AutoCreateAction.resolveTemplate(request, metadata);
+        assertThat(result, notNullValue());
+        assertThat(result.getDataStreamTemplate(), nullValue());
 
         // An index that doesn't match with an index template
         request = new CreateIndexRequest("my-index");
-        result = AutoCreateAction.resolveAutoCreateDataStream(request, metadata);
+        result = AutoCreateAction.resolveTemplate(request, metadata);
         assertThat(result, nullValue());
     }
 

server/src/test/java/org/elasticsearch/action/admin/indices/rollover/MetadataRolloverServiceTests.java

@@ -375,7 +375,7 @@ public class MetadataRolloverServiceTests extends ESTestCase {
         aliases.put("bar-write", AliasMetadata.builder("bar-write").writeIndex(randomBoolean()).build());
         final ComposableIndexTemplate template = new ComposableIndexTemplate(Arrays.asList("foo-*", "bar-*"),
             new Template(null, null, aliases),
-            null, null, null, null, null);
+            null, null, null, null, null, null);
 
         final Metadata metadata = Metadata.builder().put(createMetadata(randomAlphaOfLengthBetween(5, 7)), false)
             .put("test-template", template).build();
@@ -392,7 +392,7 @@ public class MetadataRolloverServiceTests extends ESTestCase {
         aliases.put("bar-write", AliasMetadata.builder("bar-write").writeIndex(randomBoolean()).build());
         final ComponentTemplate ct = new ComponentTemplate(new Template(null, null, aliases), null, null);
         final ComposableIndexTemplate template = new ComposableIndexTemplate(Arrays.asList("foo-*", "bar-*"), null,
-            Collections.singletonList("ct"), null, null, null, null);
+            Collections.singletonList("ct"), null, null, null, null, null);
 
         final Metadata metadata = Metadata.builder().put(createMetadata(randomAlphaOfLengthBetween(5, 7)), false)
             .put("ct", ct)
@@ -428,7 +428,7 @@ public class MetadataRolloverServiceTests extends ESTestCase {
         aliases.put("bar-write", AliasMetadata.builder("bar-write").writeIndex(randomBoolean()).build());
         final ComposableIndexTemplate template = new ComposableIndexTemplate(Collections.singletonList("*"),
             new Template(null, null, aliases),
-            null, null, null, null, null);
+            null, null, null, null, null, null);
 
         final Metadata metadata = Metadata.builder().put(createMetadata(randomAlphaOfLengthBetween(5, 7)), false)
             .put("test-template", template).build();
@@ -449,7 +449,7 @@ public class MetadataRolloverServiceTests extends ESTestCase {
         aliases.put("bar-write", AliasMetadata.builder("bar-write").writeIndex(randomBoolean()).build());
         final ComponentTemplate ct = new ComponentTemplate(new Template(null, null, aliases), null, null);
         final ComposableIndexTemplate template = new ComposableIndexTemplate(Collections.singletonList("*"), null,
-            Collections.singletonList("ct"), null, null, null, null);
+            Collections.singletonList("ct"), null, null, null, null, null);
 
         final Metadata metadata = Metadata.builder().put(createMetadata(randomAlphaOfLengthBetween(5, 7)), false)
             .put("ct", ct)
@@ -541,7 +541,7 @@ public class MetadataRolloverServiceTests extends ESTestCase {
     public void testRolloverClusterStateForDataStream() throws Exception {
         final DataStream dataStream = DataStreamTestHelper.randomInstance();
         ComposableIndexTemplate template = new ComposableIndexTemplate(List.of(dataStream.getName() + "*"), null, null, null, null, null,
-            new ComposableIndexTemplate.DataStreamTemplate());
+            new ComposableIndexTemplate.DataStreamTemplate(), null);
         Metadata.Builder builder = Metadata.builder();
         builder.put("template", template);
         for (Index index : dataStream.getIndices()) {
@@ -640,7 +640,7 @@ public class MetadataRolloverServiceTests extends ESTestCase {
             sourceIndexName = dataStream.getIndices().get(dataStream.getIndices().size() - 1).getName();
             defaultRolloverIndexName = DataStream.getDefaultBackingIndexName(dataStream.getName(), dataStream.getGeneration() + 1);
             ComposableIndexTemplate template = new ComposableIndexTemplate(List.of(dataStream.getName() + "*"), null, null, null, null,
-                null, new ComposableIndexTemplate.DataStreamTemplate());
+                null, new ComposableIndexTemplate.DataStreamTemplate(), null);
             builder.put("template", template);
             for (Index index : dataStream.getIndices()) {
                 builder.put(DataStreamTestHelper.getIndexMetadataBuilderForIndex(index));

server/src/test/java/org/elasticsearch/action/admin/indices/template/post/SimulateIndexTemplateRequestTests.java

@@ -63,7 +63,7 @@ public class SimulateIndexTemplateRequestTests extends AbstractWireSerializingTe
 
     public void testAddingGlobalTemplateWithHiddenIndexSettingIsIllegal() {
         Template template = new Template(Settings.builder().put(IndexMetadata.SETTING_INDEX_HIDDEN, true).build(), null, null);
-        ComposableIndexTemplate globalTemplate = new ComposableIndexTemplate(List.of("*"), template, null, null, null, null, null);
+        ComposableIndexTemplate globalTemplate = new ComposableIndexTemplate(List.of("*"), template, null, null, null, null, null, null);
 
         PutComposableIndexTemplateAction.Request request = new PutComposableIndexTemplateAction.Request("test");
         request.indexTemplate(globalTemplate);

server/src/test/java/org/elasticsearch/action/admin/indices/template/post/SimulateTemplateRequestTests.java

@@ -64,7 +64,7 @@ public class SimulateTemplateRequestTests extends AbstractWireSerializingTestCas
 
     public void testAddingGlobalTemplateWithHiddenIndexSettingIsIllegal() {
         Template template = new Template(Settings.builder().put(IndexMetadata.SETTING_INDEX_HIDDEN, true).build(), null, null);
-        ComposableIndexTemplate globalTemplate = new ComposableIndexTemplate(List.of("*"), template, null, null, null, null, null);
+        ComposableIndexTemplate globalTemplate = new ComposableIndexTemplate(List.of("*"), template, null, null, null, null, null, null);
 
         PutComposableIndexTemplateAction.Request request = new PutComposableIndexTemplateAction.Request("test");
         request.indexTemplate(globalTemplate);

server/src/test/java/org/elasticsearch/action/admin/indices/template/put/PutComposableIndexTemplateRequestTests.java

@@ -58,7 +58,7 @@ public class PutComposableIndexTemplateRequestTests extends AbstractWireSerializ
 
     public void testPutGlobalTemplatesCannotHaveHiddenIndexSetting() {
         Template template = new Template(Settings.builder().put(IndexMetadata.SETTING_INDEX_HIDDEN, true).build(), null, null);
-        ComposableIndexTemplate globalTemplate = new ComposableIndexTemplate(List.of("*"), template, null, null, null, null, null);
+        ComposableIndexTemplate globalTemplate = new ComposableIndexTemplate(List.of("*"), template, null, null, null, null, null, null);
 
         PutComposableIndexTemplateAction.Request request = new PutComposableIndexTemplateAction.Request("test");
         request.indexTemplate(globalTemplate);
@@ -84,7 +84,7 @@ public class PutComposableIndexTemplateRequestTests extends AbstractWireSerializ
 
     public void testValidationOfPriority() {
         PutComposableIndexTemplateAction.Request req = new PutComposableIndexTemplateAction.Request("test");
-        req.indexTemplate(new ComposableIndexTemplate(Arrays.asList("foo", "bar"), null, null, -5L, null, null, null));
+        req.indexTemplate(new ComposableIndexTemplate(Arrays.asList("foo", "bar"), null, null, -5L, null, null, null, null));
         ActionRequestValidationException validationException = req.validate();
         assertThat(validationException, is(notNullValue()));
         List<String> validationErrors = validationException.validationErrors();

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIndicesThatCannotBeCreatedTests.java

@@ -27,6 +27,7 @@ import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
@@ -35,6 +36,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexingPressure;
 import org.elasticsearch.index.VersionType;
@@ -45,34 +47,32 @@ import org.elasticsearch.test.VersionUtils;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-import java.util.Arrays;
-import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
-import java.util.function.Predicate;
+import java.util.function.Consumer;
+import java.util.function.Function;
 
 import static java.util.Collections.emptySet;
-import static java.util.Collections.singleton;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 public class TransportBulkActionIndicesThatCannotBeCreatedTests extends ESTestCase {
+    private static final Consumer<String> noop = index -> {};
+
     public void testNonExceptional() {
         BulkRequest bulkRequest = new BulkRequest();
         bulkRequest.add(new IndexRequest(randomAlphaOfLength(5)));
         bulkRequest.add(new IndexRequest(randomAlphaOfLength(5)));
         bulkRequest.add(new DeleteRequest(randomAlphaOfLength(5)));
         bulkRequest.add(new UpdateRequest(randomAlphaOfLength(5), randomAlphaOfLength(5)));
-        // Test emulating auto_create_index=false
-        indicesThatCannotBeCreatedTestCase(emptySet(), bulkRequest, null);
-        // Test emulating auto_create_index=true
-        indicesThatCannotBeCreatedTestCase(emptySet(), bulkRequest, index -> true);
-        // Test emulating all indices already created
-        indicesThatCannotBeCreatedTestCase(emptySet(), bulkRequest, index -> false);
+        // Test emulating that index can be auto-created
+        indicesThatCannotBeCreatedTestCase(emptySet(), bulkRequest, index -> true, noop);
+        // Test emulating that index cannot be auto-created
+        indicesThatCannotBeCreatedTestCase(emptySet(), bulkRequest, index -> false, noop);
         // Test emulating auto_create_index=true with some indices already created.
-        indicesThatCannotBeCreatedTestCase(emptySet(), bulkRequest, index -> randomBoolean());
+        indicesThatCannotBeCreatedTestCase(emptySet(), bulkRequest, index -> randomBoolean(), noop);
     }
 
     public void testAllFail() {
@@ -81,7 +81,7 @@ public class TransportBulkActionIndicesThatCannotBeCreatedTests extends ESTestCa
         bulkRequest.add(new IndexRequest("can't"));
         bulkRequest.add(new DeleteRequest("do").version(0).versionType(VersionType.EXTERNAL));
         bulkRequest.add(new UpdateRequest("nothin", randomAlphaOfLength(5)));
-        indicesThatCannotBeCreatedTestCase(new HashSet<>(Arrays.asList("no", "can't", "do", "nothin")), bulkRequest, index -> {
+        indicesThatCannotBeCreatedTestCase(Set.of("no", "can't", "do", "nothin"), bulkRequest, index -> true, index -> {
             throw new IndexNotFoundException("Can't make it because I say so");
         });
     }
@@ -91,40 +91,43 @@ public class TransportBulkActionIndicesThatCannotBeCreatedTests extends ESTestCa
         bulkRequest.add(new IndexRequest("ok"));
         bulkRequest.add(new IndexRequest("bad"));
         // Emulate auto_create_index=-bad,+*
-        indicesThatCannotBeCreatedTestCase(singleton("bad"), bulkRequest, index -> {
-            if (index.equals("bad")) {
-                throw new IndexNotFoundException("Can't make it because I say so");
-            }
-            return true;
-        });
-        // Emulate auto_create_index=false but the "ok" index already exists
-        indicesThatCannotBeCreatedTestCase(singleton("bad"), bulkRequest, index -> {
+        indicesThatCannotBeCreatedTestCase(Set.of("bad"), bulkRequest, index -> true, index -> {
             if (index.equals("bad")) {
                 throw new IndexNotFoundException("Can't make it because I say so");
             }
-            return false;
         });
     }
 
 
     private void indicesThatCannotBeCreatedTestCase(Set<String> expected,
-            BulkRequest bulkRequest, Predicate<String> shouldAutoCreate) {
+            BulkRequest bulkRequest, Function<String, Boolean> shouldAutoCreate, Consumer<String> simulateAutoCreate) {
         ClusterService clusterService = mock(ClusterService.class);
         ClusterState state = mock(ClusterState.class);
         when(state.getMetadata()).thenReturn(Metadata.EMPTY_METADATA);
         when(state.metadata()).thenReturn(Metadata.EMPTY_METADATA);
         when(clusterService.state()).thenReturn(state);
+
         DiscoveryNodes discoveryNodes = mock(DiscoveryNodes.class);
         when(state.getNodes()).thenReturn(discoveryNodes);
         when(discoveryNodes.getMinNodeVersion()).thenReturn(VersionUtils.randomCompatibleVersion(random(), Version.CURRENT));
+
         DiscoveryNode localNode = mock(DiscoveryNode.class);
         when(clusterService.localNode()).thenReturn(localNode);
         when(localNode.isIngestNode()).thenReturn(randomBoolean());
+
         final ThreadPool threadPool = mock(ThreadPool.class);
         final ExecutorService direct = EsExecutors.newDirectExecutorService();
         when(threadPool.executor(anyString())).thenReturn(direct);
+
+        final IndexNameExpressionResolver indexNameExpressionResolver = new IndexNameExpressionResolver(new ThreadContext(Settings.EMPTY)) {
+            @Override
+            public boolean hasIndexAbstraction(String indexAbstraction, ClusterState state) {
+                return shouldAutoCreate.apply(indexAbstraction) == false;
+            }
+        };
+
         TransportBulkAction action = new TransportBulkAction(threadPool, mock(TransportService.class), clusterService,
-                null, null, mock(ActionFilters.class), null, null,
+            null, null, mock(ActionFilters.class), indexNameExpressionResolver,
             new IndexingPressure(Settings.EMPTY), new SystemIndices(Map.of())) {
             @Override
             void executeBulk(Task task, BulkRequest bulkRequest, long startTimeNanos, ActionListener<BulkResponse> listener,
@@ -133,21 +136,25 @@ public class TransportBulkActionIndicesThatCannotBeCreatedTests extends ESTestCa
             }
 
             @Override
-            boolean needToCheck() {
-                return null != shouldAutoCreate; // Use "null" to mean "no indices can be created so don't bother checking"
+            void createIndex(String index, TimeValue timeout, Version minNodeVersion, ActionListener<CreateIndexResponse> listener) {
+                try {
+                    simulateAutoCreate.accept(index);
+                    // If we try to create an index just immediately assume it worked
+                    listener.onResponse(new CreateIndexResponse(true, true, index) {
+                    });
+                } catch (Exception e) {
+                    listener.onFailure(e);
+                }
             }
-
+        };
+        action.doExecute(null, bulkRequest, new ActionListener<>() {
             @Override
-            boolean shouldAutoCreate(String index, ClusterState state) {
-                return shouldAutoCreate.test(index);
-            }
+            public void onResponse(BulkResponse bulkItemResponses) {}
 
             @Override
-            void createIndex(String index, TimeValue timeout, Version minNodeVersion, ActionListener<CreateIndexResponse> listener) {
-                // If we try to create an index just immediately assume it worked
-                listener.onResponse(new CreateIndexResponse(true, true, index) {});
+            public void onFailure(Exception e) {
+                throw new AssertionError(e);
             }
-        };
-        action.doExecute(null, bulkRequest, null);
+        });
     }
 }

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIngestTests.java

@@ -34,10 +34,10 @@ import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateApplier;
 import org.elasticsearch.cluster.metadata.AliasMetadata;
+import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.IndexTemplateMetadata;
-import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.Template;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -45,7 +45,6 @@ import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
-import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
@@ -143,18 +142,11 @@ public class TransportBulkActionIngestTests extends ESTestCase {
 
         TestTransportBulkAction() {
             super(threadPool, transportService, clusterService, ingestService,
-                null, new ActionFilters(Collections.emptySet()), null,
-                new AutoCreateIndex(
-                    SETTINGS, new ClusterSettings(SETTINGS, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
-                    new IndexNameExpressionResolver(new ThreadContext(Settings.EMPTY)),
-                    new SystemIndices(Map.of())
-                ), new IndexingPressure(SETTINGS), new SystemIndices(Map.of())
+                null, new ActionFilters(Collections.emptySet()), new IndexNameExpressionResolver(new ThreadContext(Settings.EMPTY)),
+                new IndexingPressure(SETTINGS), new SystemIndices(Map.of())
             );
         }
-        @Override
-        protected boolean needToCheck() {
-            return needToCheck;
-        }
+
         @Override
         void executeBulk(Task task, final BulkRequest bulkRequest, final long startTimeNanos, final ActionListener<BulkResponse> listener,
                 final AtomicArray<BulkItemResponse> responses, Map<String, IndexNotFoundException> indicesThatCannotBeCreated) {
@@ -633,7 +625,7 @@ public class TransportBulkActionIngestTests extends ESTestCase {
 
         ComposableIndexTemplate t1 = new ComposableIndexTemplate(Collections.singletonList("missing_*"),
             new Template(Settings.builder().put(IndexSettings.DEFAULT_PIPELINE.getKey(), "pipeline2").build(), null, null),
-            null, null, null, null, null);
+            null, null, null, null, null, null);
 
         ClusterState state = clusterService.state();
         Metadata metadata = Metadata.builder()

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionTests.java

@@ -28,7 +28,6 @@ import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.ActionTestUtils;
-import org.elasticsearch.action.support.AutoCreateIndex;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.cluster.ClusterState;
@@ -85,15 +84,9 @@ public class TransportBulkActionTests extends ESTestCase {
         TestTransportBulkAction() {
             super(TransportBulkActionTests.this.threadPool, transportService, clusterService, null,
                     null, new ActionFilters(Collections.emptySet()), new Resolver(),
-                    new AutoCreateIndex(Settings.EMPTY, clusterService.getClusterSettings(), new Resolver(), new SystemIndices(Map.of())),
                     new IndexingPressure(Settings.EMPTY), new SystemIndices(Map.of()));
         }
 
-        @Override
-        protected boolean needToCheck() {
-            return true;
-        }
-
         @Override
         void createIndex(String index, TimeValue timeout, Version minNodeVersion, ActionListener<CreateIndexResponse> listener) {
             indexCreated = true;
@@ -268,27 +261,6 @@ public class TransportBulkActionTests extends ESTestCase {
         assertFalse(bulkAction.isOnlySystem(buildBulkRequest(mixed), indicesLookup, systemIndices));
     }
 
-    public void testIncludesSystem() {
-        SortedMap<String, IndexAbstraction> indicesLookup = new TreeMap<>();
-        Settings settings = Settings.builder().put("index.version.created", Version.CURRENT).build();
-        indicesLookup.put(".foo",
-            new Index(IndexMetadata.builder(".foo").settings(settings).system(true).numberOfShards(1).numberOfReplicas(0).build()));
-        indicesLookup.put(".bar",
-            new Index(IndexMetadata.builder(".bar").settings(settings).system(true).numberOfShards(1).numberOfReplicas(0).build()));
-        SystemIndices systemIndices = new SystemIndices(Map.of("plugin", List.of(new SystemIndexDescriptor(".test", ""))));
-        List<String> onlySystem = List.of(".foo", ".bar");
-        assertTrue(bulkAction.includesSystem(buildBulkRequest(onlySystem), indicesLookup, systemIndices));
-
-        onlySystem = List.of(".foo", ".bar", ".test");
-        assertTrue(bulkAction.includesSystem(buildBulkRequest(onlySystem), indicesLookup, systemIndices));
-
-        List<String> nonSystem = List.of("foo", "bar");
-        assertFalse(bulkAction.includesSystem(buildBulkRequest(nonSystem), indicesLookup, systemIndices));
-
-        List<String> mixed = List.of(".foo", ".test", "other");
-        assertTrue(bulkAction.includesSystem(buildBulkRequest(mixed), indicesLookup, systemIndices));
-    }
-
     private BulkRequest buildBulkRequest(List<String> indices) {
         BulkRequest request = new BulkRequest();
         for (String index : indices) {

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionTookTests.java

@@ -29,7 +29,6 @@ import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.AutoCreateIndex;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
@@ -126,7 +125,6 @@ public class TransportBulkActionTookTests extends ESTestCase {
                     client,
                     actionFilters,
                     resolver,
-                    null,
                     expected::get) {
 
                 @Override
@@ -149,7 +147,6 @@ public class TransportBulkActionTookTests extends ESTestCase {
                     client,
                     actionFilters,
                     resolver,
-                    null,
                     System::nanoTime) {
 
                 @Override
@@ -229,7 +226,6 @@ public class TransportBulkActionTookTests extends ESTestCase {
                 NodeClient client,
                 ActionFilters actionFilters,
                 IndexNameExpressionResolver indexNameExpressionResolver,
-                AutoCreateIndex autoCreateIndex,
                 LongSupplier relativeTimeProvider) {
             super(
                     threadPool,
@@ -239,21 +235,9 @@ public class TransportBulkActionTookTests extends ESTestCase {
                     client,
                     actionFilters,
                     indexNameExpressionResolver,
-                    autoCreateIndex,
                     new IndexingPressure(Settings.EMPTY),
                     new SystemIndices(Map.of()),
                     relativeTimeProvider);
         }
-
-        @Override
-        boolean needToCheck() {
-            return randomBoolean();
-        }
-
-        @Override
-        boolean shouldAutoCreate(String index, ClusterState state) {
-            return randomBoolean();
-        }
-
     }
 }

server/src/test/java/org/elasticsearch/action/support/AutoCreateIndexTests.java

@@ -22,6 +22,7 @@ package org.elasticsearch.action.support;
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.Metadata;
@@ -211,6 +212,113 @@ public class AutoCreateIndexTests extends ESTestCase {
         assertThat(autoCreateIndex.getAutoCreate().getExpressions().get(0).v1(), equalTo("logs-*"));
     }
 
+    /**
+     * Check that if a template matches the index to be created, but that template does not have a value
+     * for the allow_auto_create setting at all, and the auto_create_index setting matches the index
+     * to be created, then the null in the template does not override the auto_create_index logic and the
+     * index can be created.
+     */
+    public void testNullAllowAutoCreateInTemplateDoesNotOverrideMatchingAutoCreateIndexSetting() {
+        String randomIndex = randomAlphaOfLengthBetween(2, 10);
+        final ComposableIndexTemplate template = new ComposableIndexTemplate(
+            List.of(randomIndex.charAt(0) + "*"),
+            null,
+            List.of(),
+            null,
+            null,
+            Map.of(),
+            null,
+            null
+        );
+
+        final Metadata metadata = Metadata.builder().indexTemplates(Map.of("test_template", template)).build();
+        final ClusterState clusterState = ClusterState.builder(buildClusterState()).metadata(metadata).build();
+
+        Settings settings = Settings.builder().put(AutoCreateIndex.AUTO_CREATE_INDEX_SETTING.getKey(), randomIndex.charAt(0) + "*").build();
+        AutoCreateIndex autoCreateIndex = newAutoCreateIndex(settings);
+        assertTrue(autoCreateIndex.shouldAutoCreate(randomIndex, clusterState));
+    }
+
+    /**
+     * Check that if a template matches the index to be created, but that template does not have a value
+     * for the allow_auto_create setting at all, then it does not cause the auto-create logic to trip over
+     * on a null value.
+     */
+    public void testCanHandleNullAutoCreateSettingInTemplate() {
+        String randomIndex = randomAlphaOfLengthBetween(2, 10);
+        final ComposableIndexTemplate template = new ComposableIndexTemplate(
+            List.of(randomIndex.charAt(0) + "*"),
+            null,
+            List.of(),
+            null,
+            null,
+            Map.of(),
+            null,
+            null
+        );
+
+        final Metadata metadata = Metadata.builder().indexTemplates(Map.of("test_template", template)).build();
+        final ClusterState clusterState = ClusterState.builder(buildClusterState()).metadata(metadata).build();
+
+        Settings settings = Settings.builder().put(AutoCreateIndex.AUTO_CREATE_INDEX_SETTING.getKey(), false).build();
+        AutoCreateIndex autoCreateIndex = newAutoCreateIndex(settings);
+        IndexNotFoundException e = expectThrows(IndexNotFoundException.class, () ->
+            autoCreateIndex.shouldAutoCreate(randomIndex, clusterState));
+        assertEquals("no such index [" + randomIndex + "] and [action.auto_create_index] is [false]", e.getMessage());
+    }
+
+    /**
+     * Check that if a template matches the index to be created, but that template has the allow_auto_create
+     * setting turned off, then it overrides the global setting.
+     */
+    public void testDisabledAutoCreateTemplateSettingDoesNotOverride() {
+        String randomIndex = randomAlphaOfLengthBetween(2, 10);
+        final ComposableIndexTemplate template = new ComposableIndexTemplate(
+            List.of(randomIndex.charAt(0) + "*"),
+            null,
+            List.of(),
+            null,
+            null,
+            Map.of(),
+            null,
+            false
+        );
+
+        final Metadata metadata = Metadata.builder().indexTemplates(Map.of("test_template", template)).build();
+        final ClusterState clusterState = ClusterState.builder(buildClusterState()).metadata(metadata).build();
+
+        Settings settings = Settings.builder().put(AutoCreateIndex.AUTO_CREATE_INDEX_SETTING.getKey(), false).build();
+        AutoCreateIndex autoCreateIndex = newAutoCreateIndex(settings);
+        IndexNotFoundException e = expectThrows(IndexNotFoundException.class, () ->
+            autoCreateIndex.shouldAutoCreate(randomIndex, clusterState));
+        assertEquals("no such index [composable template [" + randomIndex.charAt(0) + "*] forbids index auto creation]", e.getMessage());
+    }
+
+    /**
+     * Check that if a template matches the index to be created, and that template has the allow_auto_create
+     * setting enabled, then it overrides the global setting.
+     */
+    public void testEnabledAutoCreateTemplateSettingDoesOverride() {
+        String randomIndex = randomAlphaOfLengthBetween(2, 10);
+        final ComposableIndexTemplate template = new ComposableIndexTemplate(
+            List.of(randomIndex.charAt(0) + "*"),
+            null,
+            List.of(),
+            null,
+            null,
+            Map.of(),
+            null,
+            true
+        );
+
+        final Metadata metadata = Metadata.builder().indexTemplates(Map.of("test_template", template)).build();
+        final ClusterState clusterState = ClusterState.builder(buildClusterState()).metadata(metadata).build();
+
+        Settings settings = Settings.builder().put(AutoCreateIndex.AUTO_CREATE_INDEX_SETTING.getKey(), false).build();
+        AutoCreateIndex autoCreateIndex = newAutoCreateIndex(settings);
+        assertThat(autoCreateIndex.shouldAutoCreate(randomIndex, clusterState), equalTo(true));
+    }
+
     private static ClusterState buildClusterState(String... indices) {
         Metadata.Builder metadata = Metadata.builder();
         for (String index : indices) {

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

@@ -97,7 +97,9 @@ public class ComposableIndexTemplateTests extends AbstractDiffableSerializationT
             randomBoolean() ? null : randomNonNegativeLong(),
             randomBoolean() ? null : randomNonNegativeLong(),
             meta,
-            dataStreamTemplate);
+            dataStreamTemplate,
+            randomBoolean() ? null : randomBoolean()
+        );
     }
 
     private static Map<String, AliasMetadata> randomAliases() {
@@ -163,7 +165,7 @@ public class ComposableIndexTemplateTests extends AbstractDiffableSerializationT
                 List<String> newIndexPatterns = randomValueOtherThan(orig.indexPatterns(),
                     () -> randomList(1, 4, () -> randomAlphaOfLength(4)));
                 return new ComposableIndexTemplate(newIndexPatterns, orig.template(), orig.composedOf(),
-                    orig.priority(), orig.version(), orig.metadata(), orig.getDataStreamTemplate());
+                    orig.priority(), orig.version(), orig.metadata(), orig.getDataStreamTemplate(), null);
             case 1:
                 return new ComposableIndexTemplate(orig.indexPatterns(),
                     randomValueOtherThan(orig.template(), () -> new Template(randomSettings(),
@@ -172,7 +174,8 @@ public class ComposableIndexTemplateTests extends AbstractDiffableSerializationT
                     orig.priority(),
                     orig.version(),
                     orig.metadata(),
-                    orig.getDataStreamTemplate());
+                    orig.getDataStreamTemplate(),
+                    orig.getAllowAutoCreate());
             case 2:
                 List<String> newComposedOf = randomValueOtherThan(orig.composedOf(),
                     () -> randomList(0, 10, () -> randomAlphaOfLength(5)));
@@ -182,7 +185,8 @@ public class ComposableIndexTemplateTests extends AbstractDiffableSerializationT
                     orig.priority(),
                     orig.version(),
                     orig.metadata(),
-                    orig.getDataStreamTemplate());
+                    orig.getDataStreamTemplate(),
+                    orig.getAllowAutoCreate());
             case 3:
                 return new ComposableIndexTemplate(orig.indexPatterns(),
                     orig.template(),
@@ -190,7 +194,8 @@ public class ComposableIndexTemplateTests extends AbstractDiffableSerializationT
                     randomValueOtherThan(orig.priority(), ESTestCase::randomNonNegativeLong),
                     orig.version(),
                     orig.metadata(),
-                    orig.getDataStreamTemplate());
+                    orig.getDataStreamTemplate(),
+                    orig.getAllowAutoCreate());
             case 4:
                 return new ComposableIndexTemplate(orig.indexPatterns(),
                     orig.template(),
@@ -198,7 +203,8 @@ public class ComposableIndexTemplateTests extends AbstractDiffableSerializationT
                     orig.priority(),
                     randomValueOtherThan(orig.version(), ESTestCase::randomNonNegativeLong),
                     orig.metadata(),
-                    orig.getDataStreamTemplate());
+                    orig.getDataStreamTemplate(),
+                    orig.getAllowAutoCreate());
             case 5:
                 return new ComposableIndexTemplate(orig.indexPatterns(),
                     orig.template(),
@@ -206,7 +212,8 @@ public class ComposableIndexTemplateTests extends AbstractDiffableSerializationT
                     orig.priority(),
                     orig.version(),
                     randomValueOtherThan(orig.metadata(), ComposableIndexTemplateTests::randomMeta),
-                    orig.getDataStreamTemplate());
+                    orig.getDataStreamTemplate(),
+                    orig.getAllowAutoCreate());
             case 6:
                 return new ComposableIndexTemplate(orig.indexPatterns(),
                     orig.template(),
@@ -214,7 +221,8 @@ public class ComposableIndexTemplateTests extends AbstractDiffableSerializationT
                     orig.priority(),
                     orig.version(),
                     orig.metadata(),
-                    randomValueOtherThan(orig.getDataStreamTemplate(), ComposableIndexTemplateTests::randomDataStreamTemplate));
+                    randomValueOtherThan(orig.getDataStreamTemplate(), ComposableIndexTemplateTests::randomDataStreamTemplate),
+                    orig.getAllowAutoCreate());
             default:
                 throw new IllegalStateException("illegal randomization branch");
         }

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

@@ -48,7 +48,7 @@ public class MetadataCreateDataStreamServiceTests extends ESTestCase {
         final MetadataCreateIndexService metadataCreateIndexService = getMetadataCreateIndexService();
         final String dataStreamName = "my-data-stream";
         ComposableIndexTemplate template = new ComposableIndexTemplate(List.of(dataStreamName + "*"), null, null, null, null, null,
-            new ComposableIndexTemplate.DataStreamTemplate());
+            new ComposableIndexTemplate.DataStreamTemplate(), null);
         ClusterState cs = ClusterState.builder(new ClusterName("_name"))
             .metadata(Metadata.builder().put("template", template).build())
             .build();
@@ -126,7 +126,8 @@ public class MetadataCreateDataStreamServiceTests extends ESTestCase {
     public void testCreateDataStreamNoValidTemplate() throws Exception {
         final MetadataCreateIndexService metadataCreateIndexService = getMetadataCreateIndexService();
         final String dataStreamName = "my-data-stream";
-        ComposableIndexTemplate template = new ComposableIndexTemplate(List.of(dataStreamName + "*"), null, null, null, null, null, null);
+        ComposableIndexTemplate template = new ComposableIndexTemplate(List.of(dataStreamName + "*"), null, null, null, null, null, null,
+            null);
         ClusterState cs = ClusterState.builder(new ClusterName("_name"))
             .metadata(Metadata.builder().put("template", template).build())
             .build();
@@ -141,7 +142,7 @@ public class MetadataCreateDataStreamServiceTests extends ESTestCase {
     public static ClusterState createDataStream(final String dataStreamName) throws Exception {
         final MetadataCreateIndexService metadataCreateIndexService = getMetadataCreateIndexService();
         ComposableIndexTemplate template = new ComposableIndexTemplate(List.of(dataStreamName + "*"), null, null, null, null, null,
-            new ComposableIndexTemplate.DataStreamTemplate());
+            new ComposableIndexTemplate.DataStreamTemplate(), null);
         ClusterState cs = ClusterState.builder(new ClusterName("_name"))
             .metadata(Metadata.builder().put("template", template).build())
             .build();

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

@@ -368,15 +368,15 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
         assertNotNull(state.metadata().componentTemplates().get("foo"));
 
         ComposableIndexTemplate firstGlobalIndexTemplate =
-            new ComposableIndexTemplate(List.of("*"), template, List.of("foo"), 1L, null, null, null);
+            new ComposableIndexTemplate(List.of("*"), template, List.of("foo"), 1L, null, null, null, null);
         state = metadataIndexTemplateService.addIndexTemplateV2(state, true, "globalindextemplate1", firstGlobalIndexTemplate);
 
         ComposableIndexTemplate secondGlobalIndexTemplate =
-            new ComposableIndexTemplate(List.of("*"), template, List.of("foo"), 2L, null, null, null);
+            new ComposableIndexTemplate(List.of("*"), template, List.of("foo"), 2L, null, null, null, null);
         state = metadataIndexTemplateService.addIndexTemplateV2(state, true, "globalindextemplate2", secondGlobalIndexTemplate);
 
         ComposableIndexTemplate fooPatternIndexTemplate =
-            new ComposableIndexTemplate(List.of("foo-*"), template, List.of("foo"), 3L, null, null, null);
+            new ComposableIndexTemplate(List.of("foo-*"), template, List.of("foo"), 3L, null, null, null, null);
         state = metadataIndexTemplateService.addIndexTemplateV2(state, true, "foopatternindextemplate", fooPatternIndexTemplate);
 
         // update the component template to set the index.hidden setting
@@ -428,7 +428,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
         List<String> patterns = new ArrayList<>(template.indexPatterns());
         patterns.add("new-pattern");
         template = new ComposableIndexTemplate(patterns, template.template(), template.composedOf(), template.priority(),
-            template.version(), template.metadata(), null);
+            template.version(), template.metadata(), null, null);
         state = metadataIndexTemplateService.addIndexTemplateV2(state, false, "foo", template);
 
         assertNotNull(state.metadata().templatesV2().get("foo"));
@@ -466,7 +466,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
             .build();
 
         ComposableIndexTemplate v2Template = new ComposableIndexTemplate(Arrays.asList("foo-bar-*", "eggplant"),
-            null, null, null, null, null, null);
+            null, null, null, null, null, null, null);
         state = metadataIndexTemplateService.addIndexTemplateV2(state, false, "v2-template", v2Template);
 
         assertWarnings("index template [v2-template] has index patterns [foo-bar-*, eggplant] matching patterns " +
@@ -503,7 +503,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
         waitToCreateComponentTemplate.await(10, TimeUnit.SECONDS);
 
         ComposableIndexTemplate globalIndexTemplate = new ComposableIndexTemplate(List.of("*"),
-            null, List.of("ct-with-index-hidden-setting"), null, null, null, null);
+            null, List.of("ct-with-index-hidden-setting"), null, null, null, null, null);
 
         expectThrows(InvalidIndexTemplateException.class, () ->
             metadataIndexTemplateService.putIndexTemplateV2("testing", true, "template-referencing-ct-with-hidden-index-setting",
@@ -528,7 +528,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
     public void testPuttingV1StarTemplateGeneratesWarning() throws Exception {
         final MetadataIndexTemplateService metadataIndexTemplateService = getMetadataIndexTemplateService();
         ComposableIndexTemplate v2Template = new ComposableIndexTemplate(Arrays.asList("foo-bar-*", "eggplant"),
-            null, null, null, null, null, null);
+            null, null, null, null, null, null, null);
         ClusterState state = metadataIndexTemplateService.addIndexTemplateV2(ClusterState.EMPTY_STATE, false, "v2-template", v2Template);
 
         MetadataIndexTemplateService.PutRequest req = new MetadataIndexTemplateService.PutRequest("cause", "v1-template");
@@ -549,7 +549,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
     public void testPuttingV1NonStarTemplateGeneratesError() throws Exception {
         final MetadataIndexTemplateService metadataIndexTemplateService = getMetadataIndexTemplateService();
         ComposableIndexTemplate v2Template = new ComposableIndexTemplate(Arrays.asList("foo-bar-*", "eggplant"),
-            null, null, null, null, null, null);
+            null, null, null, null, null, null, null);
         ClusterState state = metadataIndexTemplateService.addIndexTemplateV2(ClusterState.EMPTY_STATE, false, "v2-template", v2Template);
 
         MetadataIndexTemplateService.PutRequest req = new MetadataIndexTemplateService.PutRequest("cause", "v1-template");
@@ -583,7 +583,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
             .build();
 
         ComposableIndexTemplate v2Template = new ComposableIndexTemplate(Arrays.asList("foo-bar-*", "eggplant"),
-            null, null, null, null, null, null);
+            null, null, null, null, null, null, null);
         state = metadataIndexTemplateService.addIndexTemplateV2(state, false, "v2-template", v2Template);
 
         assertWarnings("index template [v2-template] has index patterns [foo-bar-*, eggplant] matching patterns " +
@@ -624,7 +624,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
             .build();
 
         ComposableIndexTemplate v2Template = new ComposableIndexTemplate(Arrays.asList("foo-bar-*", "eggplant"),
-            null, null, null, null, null, null);
+            null, null, null, null, null, null, null);
         state = metadataIndexTemplateService.addIndexTemplateV2(state, false, "v2-template", v2Template);
 
         assertWarnings("index template [v2-template] has index patterns [foo-bar-*, eggplant] matching patterns " +
@@ -650,11 +650,11 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
     public void testPuttingOverlappingV2Template() throws Exception {
         {
             ComposableIndexTemplate template = new ComposableIndexTemplate(Arrays.asList("egg*", "baz"),
-                null, null, 1L, null, null, null);
+                null, null, 1L, null, null, null, null);
             MetadataIndexTemplateService metadataIndexTemplateService = getMetadataIndexTemplateService();
             ClusterState state = metadataIndexTemplateService.addIndexTemplateV2(ClusterState.EMPTY_STATE, false, "foo", template);
             ComposableIndexTemplate newTemplate = new ComposableIndexTemplate(Arrays.asList("abc", "baz*"),
-                null, null, 1L, null, null, null);
+                null, null, 1L, null, null, null, null);
             IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
                 () -> metadataIndexTemplateService.addIndexTemplateV2(state, false, "foo2", newTemplate));
             assertThat(e.getMessage(), equalTo("index template [foo2] has index patterns [abc, baz*] matching patterns from existing " +
@@ -664,11 +664,11 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
 
         {
             ComposableIndexTemplate template = new ComposableIndexTemplate(Arrays.asList("egg*", "baz"),
-                null, null, null, null, null, null);
+                null, null, null, null, null, null, null);
             MetadataIndexTemplateService metadataIndexTemplateService = getMetadataIndexTemplateService();
             ClusterState state = metadataIndexTemplateService.addIndexTemplateV2(ClusterState.EMPTY_STATE, false, "foo", template);
             ComposableIndexTemplate newTemplate = new ComposableIndexTemplate(Arrays.asList("abc", "baz*"),
-                null, null, 0L, null, null, null);
+                null, null, 0L, null, null, null, null);
             IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
                 () -> metadataIndexTemplateService.addIndexTemplateV2(state, false, "foo2", newTemplate));
             assertThat(e.getMessage(), equalTo("index template [foo2] has index patterns [abc, baz*] matching patterns from existing " +
@@ -684,9 +684,9 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
 
         ComponentTemplate ct = ComponentTemplateTests.randomInstance();
         state = service.addComponentTemplate(state, true, "ct", ct);
-        ComposableIndexTemplate it = new ComposableIndexTemplate(List.of("i*"), null, List.of("ct"), null, 1L, null, null);
+        ComposableIndexTemplate it = new ComposableIndexTemplate(List.of("i*"), null, List.of("ct"), null, 1L, null, null, null);
         state = service.addIndexTemplateV2(state, true, "my-template", it);
-        ComposableIndexTemplate it2 = new ComposableIndexTemplate(List.of("in*"), null, List.of("ct"), 10L, 2L, null, null);
+        ComposableIndexTemplate it2 = new ComposableIndexTemplate(List.of("in*"), null, List.of("ct"), 10L, 2L, null, null, null);
         state = service.addIndexTemplateV2(state, true, "my-template2", it2);
 
         String result = MetadataIndexTemplateService.findV2Template(state.metadata(), "index", randomBoolean());
@@ -701,9 +701,9 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
 
         ComponentTemplate ct = ComponentTemplateTests.randomInstance();
         state = service.addComponentTemplate(state, true, "ct", ct);
-        ComposableIndexTemplate it = new ComposableIndexTemplate(List.of("i*"), null, List.of("ct"), 0L, 1L, null, null);
+        ComposableIndexTemplate it = new ComposableIndexTemplate(List.of("i*"), null, List.of("ct"), 0L, 1L, null, null, null);
         state = service.addIndexTemplateV2(state, true, "my-template", it);
-        ComposableIndexTemplate it2 = new ComposableIndexTemplate(List.of("*"), null, List.of("ct"), 10L, 2L, null, null);
+        ComposableIndexTemplate it2 = new ComposableIndexTemplate(List.of("*"), null, List.of("ct"), 10L, 2L, null, null, null);
         state = service.addIndexTemplateV2(state, true, "my-template2", it2);
 
         String result = MetadataIndexTemplateService.findV2Template(state.metadata(), "index", true);
@@ -716,7 +716,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
         try {
             // add an invalid global template that specifies the `index.hidden` setting
             ComposableIndexTemplate invalidGlobalTemplate = new ComposableIndexTemplate(List.of("*"),
-                templateWithHiddenSetting, List.of("ct"), 5L, 1L, null, null);
+                templateWithHiddenSetting, List.of("ct"), 5L, 1L, null, null, null);
             Metadata invalidGlobalTemplateMetadata = Metadata.builder().putCustom(ComposableIndexTemplateMetadata.TYPE,
                 new ComposableIndexTemplateMetadata(Map.of("invalid_global_template", invalidGlobalTemplate))).build();
 
@@ -760,7 +760,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
                     "      }\n" +
                     "    }\n" +
                     "  }"), null),
-            List.of("ct_low", "ct_high"), 0L, 1L, null, null);
+            List.of("ct_low", "ct_high"), 0L, 1L, null, null, null);
         state = service.addIndexTemplateV2(state, true, "my-template", it);
 
         List<CompressedXContent> mappings = MetadataIndexTemplateService.collectMappings(state, "my-template", "my-index",
@@ -824,7 +824,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
                     "      }\n" +
                     "    }\n" +
                     "  }"), null),
-            List.of("ct_low", "ct_high"), 0L, 1L, null, null);
+            List.of("ct_low", "ct_high"), 0L, 1L, null, null, null);
         state = service.addIndexTemplateV2(state, true, "my-template", it);
 
         List<CompressedXContent> mappings = MetadataIndexTemplateService.collectMappings(state, "my-template", "my-index",
@@ -876,7 +876,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
                         "      }\n" +
                         "    }\n" +
                         "  }"), null),
-                List.of("ct1"), 0L, 1L, null, new ComposableIndexTemplate.DataStreamTemplate());
+                List.of("ct1"), 0L, 1L, null, new ComposableIndexTemplate.DataStreamTemplate(), null);
             state = service.addIndexTemplateV2(state, true, "logs-data-stream-template", it);
 
             List<CompressedXContent> mappings = MetadataIndexTemplateService.collectMappings(state, "logs-data-stream-template",
@@ -915,7 +915,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
                         "      }\n" +
                         "    }\n" +
                         "  }"), null),
-                List.of("ct1"), 0L, 1L, null, null);
+                List.of("ct1"), 0L, 1L, null, null, null);
             state = service.addIndexTemplateV2(state, true, "timeseries-template", it);
 
             List<CompressedXContent> mappings = MetadataIndexTemplateService.collectMappings(state, "timeseries-template", "timeseries",
@@ -983,7 +983,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
 
             state = service.addComponentTemplate(state, true, "ct1", ct1);
             ComposableIndexTemplate it = new ComposableIndexTemplate(List.of("logs*"), null, List.of("ct1"), 0L, 1L, null,
-                new ComposableIndexTemplate.DataStreamTemplate());
+                new ComposableIndexTemplate.DataStreamTemplate(), null);
             state = service.addIndexTemplateV2(state, true, "logs-template", it);
 
             List<CompressedXContent> mappings = MetadataIndexTemplateService.collectMappings(state, "logs-template",
@@ -1018,7 +1018,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
                 "      }\n" +
                 "    }"), null);
             ComposableIndexTemplate it = new ComposableIndexTemplate(List.of("timeseries*"), template, null, 0L, 1L, null,
-                new ComposableIndexTemplate.DataStreamTemplate());
+                new ComposableIndexTemplate.DataStreamTemplate(), null);
             state = service.addIndexTemplateV2(state, true, "timeseries-template", it);
 
             List<CompressedXContent> mappings = MetadataIndexTemplateService.collectMappings(state, "timeseries-template",
@@ -1065,7 +1065,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
                 .put("index.blocks.write", false)
                 .put("index.number_of_shards", 3)
                 .build(), null, null),
-            List.of("ct_low", "ct_high"), 0L, 1L, null, null);
+            List.of("ct_low", "ct_high"), 0L, 1L, null, null, null);
         state = service.addIndexTemplateV2(state, true, "my-template", it);
 
         Settings settings = MetadataIndexTemplateService.resolveSettings(state.metadata(), "my-template");
@@ -1093,7 +1093,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
         state = service.addComponentTemplate(state, true, "ct_low", ct2);
         ComposableIndexTemplate it = new ComposableIndexTemplate(List.of("i*"),
             new Template(null, null, a3),
-            List.of("ct_low", "ct_high"), 0L, 1L, null, null);
+            List.of("ct_low", "ct_high"), 0L, 1L, null, null, null);
         state = service.addIndexTemplateV2(state, true, "my-template", it);
 
         List<Map<String, AliasMetadata>> resolvedAliases = MetadataIndexTemplateService.resolveAliases(state.metadata(), "my-template");
@@ -1209,7 +1209,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
                 "        }\n" +
                 "      }\n" +
                 "    }"), null),
-            randomBoolean() ? Arrays.asList("c1", "c2") : Arrays.asList("c2", "c1"), 0L, 1L, null, null);
+            randomBoolean() ? Arrays.asList("c1", "c2") : Arrays.asList("c2", "c1"), 0L, 1L, null, null, null);
 
         final ClusterState finalState = state;
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
@@ -1260,7 +1260,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
         state = service.addComponentTemplate(state, true, "c2", ct2);
         ComposableIndexTemplate it = new ComposableIndexTemplate(List.of("i*"),
             new Template(null, null, null),
-            randomBoolean() ? Arrays.asList("c1", "c2") : Arrays.asList("c2", "c1"), 0L, 1L, null, null);
+            randomBoolean() ? Arrays.asList("c1", "c2") : Arrays.asList("c2", "c1"), 0L, 1L, null, null, null);
 
         // Great, the templates aren't invalid
         state = service.addIndexTemplateV2(state, randomBoolean(), "my-template", it);
@@ -1332,7 +1332,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
             .build();
 
         ComposableIndexTemplate template = new ComposableIndexTemplate(Collections.singletonList("logs-*-*"),
-            null, null, 100L, null, null, new ComposableIndexTemplate.DataStreamTemplate());
+            null, null, 100L, null, null, new ComposableIndexTemplate.DataStreamTemplate(), null);
 
         state = service.addIndexTemplateV2(state, false, "logs", template);
 
@@ -1354,7 +1354,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
         // Test replacing it with a version without the data stream config
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> {
             ComposableIndexTemplate nonDSTemplate = new ComposableIndexTemplate(Collections.singletonList("logs-*-*"), null, null,
-                100L, null, null, null);
+                100L, null, null, null, null);
             service.addIndexTemplateV2(stateWithDS, false, "logs", nonDSTemplate);
         });
 
@@ -1365,7 +1365,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
         // Test adding a higher priority version that would cause problems
         e = expectThrows(IllegalArgumentException.class, () -> {
             ComposableIndexTemplate nonDSTemplate = new ComposableIndexTemplate(Collections.singletonList("logs-my*-*"), null, null,
-                105L, null, null, null);
+                105L, null, null, null, null);
             service.addIndexTemplateV2(stateWithDS, false, "logs2", nonDSTemplate);
         });
 
@@ -1376,7 +1376,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
         // Change the pattern to one that doesn't match the data stream
         e = expectThrows(IllegalArgumentException.class, () -> {
             ComposableIndexTemplate newTemplate = new ComposableIndexTemplate(Collections.singletonList("logs-postgres-*"), null,
-                null, 100L, null, null, new ComposableIndexTemplate.DataStreamTemplate());
+                null, 100L, null, null, new ComposableIndexTemplate.DataStreamTemplate(), null);
             service.addIndexTemplateV2(stateWithDS, false, "logs", newTemplate);
         });
 
@@ -1386,12 +1386,12 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
 
         // Add an additional template that matches our data stream at a lower priority
         ComposableIndexTemplate mysqlTemplate = new ComposableIndexTemplate(Collections.singletonList("logs-mysql-*"), null,
-            null, 50L, null, null, new ComposableIndexTemplate.DataStreamTemplate());
+            null, 50L, null, null, new ComposableIndexTemplate.DataStreamTemplate(), null);
         ClusterState stateWithDSAndTemplate = service.addIndexTemplateV2(stateWithDS, false, "logs-mysql", mysqlTemplate);
 
         // We should be able to replace the "logs" template, because we have the "logs-mysql" template that can handle the data stream
         ComposableIndexTemplate nonDSTemplate = new ComposableIndexTemplate(Collections.singletonList("logs-postgres-*"), null, null,
-            100L, null, null, null);
+            100L, null, null, null, null);
         service.addIndexTemplateV2(stateWithDSAndTemplate, false, "logs", nonDSTemplate);
     }
 
@@ -1475,9 +1475,9 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
     @SuppressWarnings("unchecked")
     public static void assertTemplatesEqual(ComposableIndexTemplate actual, ComposableIndexTemplate expected) {
         ComposableIndexTemplate actualNoTemplate = new ComposableIndexTemplate(actual.indexPatterns(), null,
-            actual.composedOf(), actual.priority(), actual.version(), actual.metadata(), actual.getDataStreamTemplate());
+            actual.composedOf(), actual.priority(), actual.version(), actual.metadata(), actual.getDataStreamTemplate(), null);
         ComposableIndexTemplate expectedNoTemplate = new ComposableIndexTemplate(expected.indexPatterns(), null,
-            expected.composedOf(), expected.priority(), expected.version(), expected.metadata(), expected.getDataStreamTemplate());
+            expected.composedOf(), expected.priority(), expected.version(), expected.metadata(), expected.getDataStreamTemplate(), null);
 
         assertThat(actualNoTemplate, equalTo(expectedNoTemplate));
         Template actualTemplate = actual.template();

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

@@ -79,7 +79,8 @@ public class ToAndFromJsonMetadataTests extends ESTestCase {
                     5L,
                     4L,
                     Collections.singletonMap("my_meta", Collections.singletonMap("potato", "chicken")),
-                    randomBoolean() ? null : new ComposableIndexTemplate.DataStreamTemplate()))
+                    randomBoolean() ? null : new ComposableIndexTemplate.DataStreamTemplate(),
+                    null))
                 .put(IndexMetadata.builder("test12")
                         .settings(settings(Version.CURRENT)
                                 .put("setting1", "value1")

server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java

@@ -81,7 +81,6 @@ import org.elasticsearch.action.search.TransportSearchAction;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.ActionTestUtils;
 import org.elasticsearch.action.support.ActiveShardCount;
-import org.elasticsearch.action.support.AutoCreateIndex;
 import org.elasticsearch.action.support.DestructiveOperations;
 import org.elasticsearch.action.support.GroupedActionListener;
 import org.elasticsearch.action.support.PlainActionFuture;
@@ -1587,7 +1586,6 @@ public class SnapshotResiliencyTests extends ESTestCase {
                             new AnalysisModule(environment, Collections.emptyList()).getAnalysisRegistry(),
                             Collections.emptyList(), client),
                         client, actionFilters, indexNameExpressionResolver,
-                        new AutoCreateIndex(settings, clusterSettings, indexNameExpressionResolver, new SystemIndices(Map.of())),
                         new IndexingPressure(settings),
                         new SystemIndices(Map.of())
                     ));

x-pack/plugin/core/src/internalClusterTest/java/org/elasticsearch/xpack/cluster/routing/allocation/DataTierIT.java

@@ -168,7 +168,7 @@ public class DataTierIT extends ESIntegTestCase {
         Template t = new Template(Settings.builder()
             .put(DataTierAllocationDecider.INDEX_ROUTING_REQUIRE, DataTier.DATA_WARM)
             .build(), null, null);
-        ComposableIndexTemplate ct = new ComposableIndexTemplate(Collections.singletonList(index), t, null, null, null, null, null);
+        ComposableIndexTemplate ct = new ComposableIndexTemplate(Collections.singletonList(index), t, null, null, null, null, null, null);
         client().execute(PutComposableIndexTemplateAction.INSTANCE,
             new PutComposableIndexTemplateAction.Request("template").indexTemplate(ct)).actionGet();
 
@@ -185,7 +185,7 @@ public class DataTierIT extends ESIntegTestCase {
         t = new Template(Settings.builder()
             .putNull(DataTierAllocationDecider.INDEX_ROUTING_PREFER)
             .build(), null, null);
-        ct = new ComposableIndexTemplate(Collections.singletonList(index), t, null, null, null, null, null);
+        ct = new ComposableIndexTemplate(Collections.singletonList(index), t, null, null, null, null, null, null);
         client().execute(PutComposableIndexTemplateAction.INSTANCE,
             new PutComposableIndexTemplateAction.Request("template").indexTemplate(ct)).actionGet();
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/slm/history/SnapshotLifecycleTemplateRegistry.java

@@ -38,7 +38,8 @@ public class SnapshotLifecycleTemplateRegistry extends IndexTemplateRegistry {
     // version 2: converted to hidden index
     // version 3: templates moved to composable templates
     // version 4:converted data stream
-    public static final int INDEX_TEMPLATE_VERSION = 4;
+    // version 5: add `allow_auto_create` setting
+    public static final int INDEX_TEMPLATE_VERSION = 5;
 
     public static final String SLM_TEMPLATE_VERSION_VARIABLE = "xpack.slm.template.version";
     public static final String SLM_TEMPLATE_NAME = ".slm-history";

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/watcher/support/WatcherIndexTemplateRegistryField.java

@@ -17,8 +17,9 @@ public final class WatcherIndexTemplateRegistryField {
     // version 10: add support for foreach path in actions
     // version 11: watch history indices are hidden
     // version 12: templates changed to composable templates
+    // version 13: add `allow_auto_create` setting
     // Note: if you change this, also inform the kibana team around the watcher-ui
-    public static final int INDEX_TEMPLATE_VERSION = 12;
+    public static final int INDEX_TEMPLATE_VERSION = 13;
     public static final String HISTORY_TEMPLATE_NAME = ".watch-history-" + INDEX_TEMPLATE_VERSION;
     public static final String HISTORY_TEMPLATE_NAME_NO_ILM = ".watch-history-no-ilm-" + INDEX_TEMPLATE_VERSION;
     public static final String TRIGGERED_TEMPLATE_NAME = ".triggered_watches";

x-pack/plugin/core/src/main/resources/ilm-history.json

@@ -80,6 +80,7 @@
         }
       }
   },
+  "allow_auto_create": true,
   "_meta": {
     "description": "index template for ILM history indices",
     "managed": true

x-pack/plugin/core/src/main/resources/logs-template.json

@@ -6,6 +6,7 @@
     "logs-mappings",
     "logs-settings"
   ],
+  "allow_auto_create": true,
   "_meta": {
     "description": "default logs template installed by x-pack",
     "managed": true

x-pack/plugin/core/src/main/resources/logstash-management.json

@@ -7,6 +7,7 @@
       "codec": "best_compression"
     }
   },
+  "allow_auto_create": true,
   "mappings" : {
     "_doc" : {
       "_meta": {

x-pack/plugin/core/src/main/resources/metrics-template.json

@@ -6,6 +6,7 @@
     "metrics-mappings",
     "metrics-settings"
   ],
+  "allow_auto_create": true,
   "_meta": {
     "description": "default metrics template installed by x-pack",
     "managed": true

x-pack/plugin/core/src/main/resources/slm-history.json

@@ -55,6 +55,7 @@
       }
     }
   },
+  "allow_auto_create": true,
   "_meta": {
     "description": "index template for SLM history indices",
     "managed": true

x-pack/plugin/core/src/main/resources/synthetics-template.json

@@ -6,6 +6,7 @@
     "synthetics-mappings",
     "synthetics-settings"
   ],
+  "allow_auto_create": true,
   "_meta": {
     "description": "default synthetics template installed by x-pack",
     "managed": true

x-pack/plugin/core/src/main/resources/triggered-watches.json

@@ -37,6 +37,7 @@
       }
     }
   },
+  "allow_auto_create": true,
   "_meta": {
     "description": "index template for triggered watches indices",
     "managed": true

x-pack/plugin/core/src/main/resources/watch-history-no-ilm.json

@@ -613,6 +613,7 @@
       }
     }
   },
+  "allow_auto_create": true,
   "_meta": {
     "description": "index template for watcher history indices",
     "managed": true

x-pack/plugin/core/src/main/resources/watch-history.json

@@ -564,6 +564,7 @@
       }
     }
   },
+  "allow_auto_create": true,
   "_meta": {
     "description": "index template for watcher history indices",
     "managed": true

x-pack/plugin/core/src/main/resources/watches.json

@@ -59,6 +59,7 @@
         }
     }
   },
+  "allow_auto_create": true,
   "_meta": {
     "description": "index template for watches indices",
     "managed": true

x-pack/plugin/data-streams/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/datastreams/AutoCreateDataStreamIT.java

@@ -0,0 +1,118 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.xpack.datastreams;
+
+import org.elasticsearch.action.support.AutoCreateIndex;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.Streams;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.test.rest.ESRestTestCase;
+
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.time.Instant;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.hamcrest.Matchers.containsString;
+
+public class AutoCreateDataStreamIT extends ESRestTestCase {
+
+    /**
+     * Check that setting {@link AutoCreateIndex#AUTO_CREATE_INDEX_SETTING} to <code>false</code>
+     * does not affect the ability to auto-create data streams, which are not subject to the setting.
+     */
+    public void testCanAutoCreateDataStreamWhenAutoCreateIndexDisabled() throws IOException {
+        configureAutoCreateIndex(false);
+        createTemplateWithAllowAutoCreate(null);
+        assertOK(this.indexDocument());
+    }
+
+    /**
+     * Check that automatically creating a data stream is allowed when the index name matches a template
+     * and that template has <code>allow_auto_create</code> set to <code>true</code>.
+     */
+    public void testCanAutoCreateDataStreamWhenExplicitlyAllowedByTemplate() throws IOException {
+        createTemplateWithAllowAutoCreate(true);
+
+        // Attempt to add a document to a non-existing index. Auto-creating the index should succeed because the index name
+        // matches the template pattern
+        assertOK(this.indexDocument());
+    }
+
+    /**
+     * Check that automatically creating a data stream is disallowed when the data stream name matches a template and that template has
+     * <code>allow_auto_create</code> explicitly to <code>false</code>.
+     */
+    public void testCannotAutoCreateDataStreamWhenDisallowedByTemplate() throws IOException {
+        createTemplateWithAllowAutoCreate(false);
+
+        // Attempt to add a document to a non-existing index. Auto-creating the index should succeed because the index name
+        // matches the template pattern
+        final ResponseException responseException = expectThrows(ResponseException.class, this::indexDocument);
+
+        assertThat(
+            Streams.copyToString(new InputStreamReader(responseException.getResponse().getEntity().getContent(), UTF_8)),
+            containsString("no such index [composable template [recipe*] forbids index auto creation]")
+        );
+    }
+
+    private void configureAutoCreateIndex(boolean value) throws IOException {
+        XContentBuilder builder = JsonXContent.contentBuilder()
+            .startObject()
+            .startObject("transient")
+            .field(AutoCreateIndex.AUTO_CREATE_INDEX_SETTING.getKey(), value)
+            .endObject()
+            .endObject();
+
+        final Request settingsRequest = new Request("PUT", "_cluster/settings");
+        settingsRequest.setJsonEntity(Strings.toString(builder));
+        final Response settingsResponse = client().performRequest(settingsRequest);
+        assertOK(settingsResponse);
+    }
+
+    private void createTemplateWithAllowAutoCreate(Boolean allowAutoCreate) throws IOException {
+        XContentBuilder b = JsonXContent.contentBuilder();
+        b.startObject();
+        {
+            b.array("index_patterns", "recipe*");
+            if (allowAutoCreate != null) {
+                b.field("allow_auto_create", allowAutoCreate);
+            }
+            b.startObject("data_stream");
+            b.endObject();
+        }
+        b.endObject();
+
+        final Request createTemplateRequest = new Request("PUT", "_index_template/recipe_template");
+        createTemplateRequest.setJsonEntity(Strings.toString(b));
+        final Response createTemplateResponse = client().performRequest(createTemplateRequest);
+        assertOK(createTemplateResponse);
+    }
+
+    private Response indexDocument() throws IOException {
+        final Request indexDocumentRequest = new Request("POST", "recipe_kr/_doc");
+        indexDocumentRequest.setJsonEntity("{ \"@timestamp\": \"" + Instant.now() + "\", \"name\": \"Kimchi\" }");
+        return client().performRequest(indexDocumentRequest);
+    }
+}

x-pack/plugin/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamIT.java

@@ -426,7 +426,8 @@ public class DataStreamIT extends ESIntegTestCase {
                 null,
                 null,
                 null,
-                new ComposableIndexTemplate.DataStreamTemplate()
+                new ComposableIndexTemplate.DataStreamTemplate(),
+                null
             )
         );
         client().execute(PutComposableIndexTemplateAction.INSTANCE, request).actionGet();
@@ -509,7 +510,8 @@ public class DataStreamIT extends ESIntegTestCase {
                 null,
                 null,
                 null,
-                new ComposableIndexTemplate.DataStreamTemplate()
+                new ComposableIndexTemplate.DataStreamTemplate(),
+                null
             )
         );
 
@@ -968,7 +970,8 @@ public class DataStreamIT extends ESIntegTestCase {
                 null,
                 null,
                 null,
-                new ComposableIndexTemplate.DataStreamTemplate()
+                new ComposableIndexTemplate.DataStreamTemplate(),
+                null
             )
         );
         client().execute(PutComposableIndexTemplateAction.INSTANCE, createTemplateRequest).actionGet();
@@ -1197,7 +1200,8 @@ public class DataStreamIT extends ESIntegTestCase {
                 null,
                 null,
                 null,
-                new ComposableIndexTemplate.DataStreamTemplate()
+                new ComposableIndexTemplate.DataStreamTemplate(),
+                null
             )
         );
         client().execute(PutComposableIndexTemplateAction.INSTANCE, request).actionGet();

x-pack/plugin/data-streams/src/internalClusterTest/java/org/elasticsearch/xpack/datastreams/DataTierDataStreamIT.java

@@ -43,7 +43,8 @@ public class DataTierDataStreamIT extends ESIntegTestCase {
             null,
             null,
             null,
-            new ComposableIndexTemplate.DataStreamTemplate()
+            new ComposableIndexTemplate.DataStreamTemplate(),
+            null
         );
         client().execute(
             PutComposableIndexTemplateAction.INSTANCE,

x-pack/plugin/data-streams/src/test/java/org/elasticsearch/xpack/datastreams/DataStreamsStatsTests.java

@@ -218,7 +218,8 @@ public class DataStreamsStatsTests extends ESSingleNodeTestCase {
             null,
             null,
             null,
-            new ComposableIndexTemplate.DataStreamTemplate()
+            new ComposableIndexTemplate.DataStreamTemplate(),
+            null
         );
         assertTrue(
             client().execute(

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/history/ILMHistoryTemplateRegistry.java

@@ -29,7 +29,8 @@ public class ILMHistoryTemplateRegistry extends IndexTemplateRegistry {
     // version 1: initial
     // version 2: convert to hidden index
     // version 3: templates moved to composable templates
-    public static final int INDEX_TEMPLATE_VERSION = 3;
+    // version 4: add `allow_auto_create` setting
+    public static final int INDEX_TEMPLATE_VERSION = 4;
 
     public static final String ILM_TEMPLATE_VERSION_VARIABLE = "xpack.ilm_history.template.version";
     public static final String ILM_TEMPLATE_NAME = "ilm-history";

x-pack/plugin/ml/qa/native-multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/MlNativeIntegTestCase.java

@@ -313,7 +313,8 @@ abstract class MlNativeIntegTestCase extends ESIntegTestCase {
                     null,
                     null,
                     null,
-                    new ComposableIndexTemplate.DataStreamTemplate())))
+                    new ComposableIndexTemplate.DataStreamTemplate(),
+                    null)))
             .actionGet();
         client().execute(CreateDataStreamAction.INSTANCE, new CreateDataStreamAction.Request(dataStreamName)).actionGet();
     }

x-pack/plugin/watcher/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/watcher/put_watch/90_auto_create_index.yml

@@ -0,0 +1,52 @@
+---
+setup:
+  - do:
+      cluster.health:
+          wait_for_status: yellow
+  - do:
+      cluster.put_settings:
+        body:
+          transient:
+            action.auto_create_index: false
+
+---
+teardown:
+  - do:
+      watcher.delete_watch:
+        id: "test_watch"
+        ignore: 404
+  - do:
+      cluster.put_settings:
+        body:
+          transient:
+            action.auto_create_index: null
+
+---
+"Ensure watch is created even when auto_create_index is disabled":
+  - do:
+      watcher.put_watch:
+        id: "test_watch"
+        body:  >
+          {
+            "trigger": {
+              "schedule" : { "cron" : "0 0 0 1 * ? 2099" }
+            },
+            "input": {
+              "simple": {
+                "foo": "bar"
+              }
+            },
+            "condition": {
+              "always": {}
+            },
+            "actions": {
+              "indexme" : {
+                "index" : {
+                  "index" : "my_test_index",
+                  "doc_id": "my-id"
+                }
+              }
+            }
+          }
+  - match: { _id: "test_watch" }
+  - match: { created: true }