Fail with correct error if first backing index exists when auto creating data stream (#62825)

Today if a data stream is auto created, but an index with same name as the
first backing index already exists then internally that error is ignored,
which then result that later in the execution of a bulk request, the
bulk item fails due to that the data stream hasn't been auto created.

This situation can only occur if an index with same is created that
will be the backing index of a data stream prior to the creation
of the data stream.

Co-authored-by: Dan Hermann <danhermann@users.noreply.github.com>

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

@@ -20,6 +20,7 @@ package org.elasticsearch.cluster.metadata;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.ResourceAlreadyExistsException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.create.CreateIndexClusterStateUpdateRequest;
@@ -38,6 +39,7 @@ import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.ObjectPath;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.MetadataFieldMapper;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.IOException;
@@ -143,7 +145,15 @@ public class MetadataCreateDataStreamService {
             new CreateIndexClusterStateUpdateRequest("initialize_data_stream", firstBackingIndexName, firstBackingIndexName)
                 .dataStreamName(request.name)
                 .settings(Settings.builder().put("index.hidden", true).build());
-        currentState = metadataCreateIndexService.applyCreateIndexRequest(currentState, createIndexRequest, false);
+        try {
+            currentState = metadataCreateIndexService.applyCreateIndexRequest(currentState, createIndexRequest, false);
+        } catch (ResourceAlreadyExistsException e) {
+            // Rethrow as ElasticsearchStatusException, so that bulk transport action doesn't ignore it during
+            // auto index/data stream creation.
+            // (otherwise bulk execution fails later, because data stream will also not have been created)
+            throw new ElasticsearchStatusException("data stream could not be created because backing index [{}] already exists",
+                RestStatus.BAD_REQUEST, e, firstBackingIndexName);
+        }
         IndexMetadata firstBackingIndex = currentState.metadata().index(firstBackingIndexName);
         assert firstBackingIndex != null;
         assert firstBackingIndex.mapping() != null : "no mapping found for backing index [" + firstBackingIndexName + "]";

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

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.datastreams;
 
+import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.action.DocWriteRequest;
@@ -1044,6 +1045,54 @@ public class DataStreamIT extends ESIntegTestCase {
         assertThat(getIndexResponse.getSettings().get("logs-foobar").get(IndexMetadata.SETTING_NUMBER_OF_REPLICAS), equalTo("0"));
     }
 
+    public void testCreatingDataStreamAndFirstBackingIndexExistsFails() throws Exception {
+        String dataStreamName = "logs-foobar";
+        String backingIndex = DataStream.getDefaultBackingIndexName(dataStreamName, 1);
+
+        createIndex(backingIndex);
+        putComposableIndexTemplate("id", List.of("logs-*"));
+        CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
+        Exception e = expectThrows(
+            ElasticsearchStatusException.class,
+            () -> client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).actionGet()
+        );
+        assertThat(e.getMessage(), equalTo("data stream could not be created because backing index [" + backingIndex + "] already exists"));
+    }
+
+    public void testAutoCreatingDataStreamAndFirstBackingIndexExistsFails() throws Exception {
+        String dataStreamName = "logs-foobar";
+        String backingIndex = DataStream.getDefaultBackingIndexName(dataStreamName, 1);
+
+        createIndex(backingIndex);
+        putComposableIndexTemplate("id", List.of("logs-*"));
+
+        IndexRequest indexRequest = new IndexRequest(dataStreamName).opType("create")
+            .source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON);
+        Exception e = expectThrows(ElasticsearchStatusException.class, () -> client().index(indexRequest).actionGet());
+        assertThat(e.getMessage(), equalTo("data stream could not be created because backing index [" + backingIndex + "] already exists"));
+    }
+
+    public void testCreatingDataStreamAndBackingIndexExistsFails() throws Exception {
+        String dataStreamName = "logs-foobar";
+        String backingIndex = DataStream.getDefaultBackingIndexName(dataStreamName, 2);
+
+        createIndex(backingIndex);
+        putComposableIndexTemplate("id", List.of("logs-*"));
+
+        CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
+        Exception e = expectThrows(
+            IllegalStateException.class,
+            () -> client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).actionGet()
+        );
+        assertThat(
+            e.getMessage(),
+            equalTo(
+                "data stream [logs-foobar] could create backing indices that conflict with 1 "
+                    + "existing index(s) or alias(s) including '.ds-logs-foobar-000002'"
+            )
+        );
+    }
+
     private static void verifyResolvability(String dataStream, ActionRequestBuilder<?, ?> requestBuilder, boolean fail) {
         verifyResolvability(dataStream, requestBuilder, fail, 0);
     }