Fix _bulk response when it can't create an index

[nik9000]

Before #22488 when an index couldn't be created during a _bulk
operation we'd do all the other actions and return the index
creation error on each failing action. In #22488 we accidentally
changed it so that we now reject the entire bulk request if a single
action cannot create an index that it must create to run. This
reverts to the old behavior while still keeping the nicer
error messages. Instead of failing the entire request we now only
fail the portions of the request that can't work because the index
doesn't exist.

Closes #24028

[jasontedor]

Thanks @nik9000, I did a first round. I get the change, it looks good. I offered a suggestion that might simplify things, I have a complaint about a name. I appreciate the REST test a lot, but I would like to ask if it would be possible to write a unit test. Maybe it would be possible override executeBulk and assert on the names of indices that can not be created that is passed in?

[jasontedor]

I think this can be simplified, no need to iterate and remove, and no need for the final map:

diff --git a/core/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java b/core/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java
index 1b1f961a12..708f8eb4b0 100644
--- a/core/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java
+++ b/core/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java
@@ -64,6 +64,7 @@ import org.elasticsearch.transport.TransportService;
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -142,34 +143,30 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
         if (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 Set<String> autoCreateIndices = bulkRequest.requests.stream()
+            final Set<String> indices = bulkRequest.requests.stream()
                 .map(DocWriteRequest::index)
                 .collect(Collectors.toSet());
             /* 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. */
-            Map<String, IndexNotFoundException> indicesThatCannotBeCreated = null;
+            final Map<String, IndexNotFoundException> indicesThatCannotBeCreated = new HashMap<>();
+            final Set<String> autoCreateIndices = new HashSet<>();
             ClusterState state = clusterService.state();
-            for (Iterator<String> itr = autoCreateIndices.iterator(); itr.hasNext();) {
-                String index = itr.next();
+            for (final String index : indices) {
                 boolean shouldAutoCreate;
                 try {
                     shouldAutoCreate = shouldAutoCreate(index, state);
                 } catch (IndexNotFoundException e) {
                     shouldAutoCreate = false;
-                    if (indicesThatCannotBeCreated == null) {
-                        indicesThatCannotBeCreated = new HashMap<>();
-                    }
                     indicesThatCannotBeCreated.put(index, e);
                 }
-                if (false == shouldAutoCreate) {
-                    itr.remove();
+                if (shouldAutoCreate) {
+                    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.
-            final Map<String, IndexNotFoundException> finalIndicesThatCannotBeCreated =
-                    indicesThatCannotBeCreated == null ? emptyMap() : indicesThatCannotBeCreated;
             if (autoCreateIndices.isEmpty()) {
-                executeBulk(task, bulkRequest, startTime, listener, responses, finalIndicesThatCannotBeCreated);
+                executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated);
             } else {
                 final AtomicInteger counter = new AtomicInteger(autoCreateIndices.size());
                 for (String index : autoCreateIndices) {
@@ -181,7 +178,7 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
                         @Override
                         public void onResponse(CreateIndexResponse result) {
                             if (counter.decrementAndGet() == 0) {
-                                executeBulk(task, bulkRequest, startTime, listener, responses, finalIndicesThatCannotBeCreated);
+                                executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated);
                             }
                         }
 
@@ -200,7 +197,7 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
                                 executeBulk(task, bulkRequest, startTime, ActionListener.wrap(listener::onResponse, inner -> {
                                     inner.addSuppressed(e);
                                     listener.onFailure(inner);
-                                }), responses, finalIndicesThatCannotBeCreated);
+                                }), responses, indicesThatCannotBeCreated);
                             }
                         }
                     });

[nik9000]

I didn't want to add the extra HashMap allocation. I'm happy to do it your way if you think it isn't a big deal.

[jasontedor]

I see. I think it will die young so it's probably not a big deal. I'm fine if you think it's important yet I think either way the rest of my suggestion (to avoid the iterate/remove) stand?

[jasontedor]

Of course I have a complaint about the naming. 😇

I think that we should stop calling these "smoke tests" as they are not smoke tests, and I would like to see some extra hyphens in autocreateindex.

[nik9000]

++

[nik9000]

would be possible to write a unit test.

Almost certainly

[nik9000]

@jasontedor I'm fine with the young garbage argument. I pushed some commits which should address your comments.

[jasontedor]

Double negative. 😄

[jasontedor]

I still want hyphens: auto-create-index. 😇

[nik9000]

Sure.

[jasontedor]

LGTM. I left two comments that you can address as you see fit.

[nik9000]

Thanks @jasontedor! I'll backport now.

[nik9000]

All backported except 5.x. Still waiting on that build.

[nik9000]

Backported to 5.x as well.