Fix Concurrent Index Auto Create Failing Bulk Requests (#82541)

Batching these requests introduced a bug where auto-create requests for system
indices would fail because system indices are always auto-created and thus
throw resource-already-exists if multiple equal ones are batched together
even though the index doesn't yet exist in the cluster state but only
in the intermediary task executor state.
This leads to bulk requests ignoring the exeception (thinking that the index
already exists) in their auto-create callback when the request doesn't yet
exist.
Fixed by deduplicating these requests for now, added a TODO to do it a little
nicer down the road but this fix is somewhat urgent as it breaks ML integ
tests.

Author: original-brownbear

server/src/internalClusterTest/java/org/elasticsearch/action/admin/indices/create/CreateSystemIndicesIT.java

@@ -8,6 +8,7 @@
 
 package org.elasticsearch.action.admin.indices.create;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.alias.Alias;
 import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
 import org.elasticsearch.action.admin.indices.alias.get.GetAliasesResponse;
@@ -17,7 +18,11 @@
 import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
 import org.elasticsearch.action.admin.indices.template.delete.DeleteComposableIndexTemplateAction;
 import org.elasticsearch.action.admin.indices.template.put.PutComposableIndexTemplateAction;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.metadata.AliasMetadata;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.MappingMetadata;
@@ -40,6 +45,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.indices.TestSystemIndexDescriptor.INDEX_NAME;
@@ -201,6 +208,31 @@ private void doCreateTest(Runnable runnable, String concreteIndex) {
         assertAliases(concreteIndex);
     }
 
+    public void testConcurrentAutoCreates() throws InterruptedException {
+        internalCluster().startNodes(3);
+
+        final Client client = client();
+        final int count = randomIntBetween(5, 30);
+        final CountDownLatch latch = new CountDownLatch(count);
+        final ActionListener<BulkResponse> listener = new ActionListener<>() {
+            @Override
+            public void onResponse(BulkResponse o) {
+                latch.countDown();
+                assertFalse(o.hasFailures());
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                latch.countDown();
+                throw new AssertionError(e);
+            }
+        };
+        for (int i = 0; i < count; i++) {
+            client.bulk(new BulkRequest().add(new IndexRequest(INDEX_NAME).source(Map.of("foo", "bar"))), listener);
+        }
+        assertTrue(latch.await(30L, TimeUnit.SECONDS));
+    }
+
     /**
      * Make sure that aliases are created hidden
      */