Optimize error handling after lazy rollovers (#111572)

This commit improves the performance of the error-handling process after a lazy rollover or an index creation failed.

Author: Kushal-Dalasaniya

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

@@ -87,7 +87,6 @@ final class BulkOperation extends ActionRunnable<BulkResponse> {
     private final ConcurrentLinkedQueue<BulkItemRequest> failureStoreRedirects = new ConcurrentLinkedQueue<>();
     private final long startTimeNanos;
     private final ClusterStateObserver observer;
-    private final Map<String, IndexNotFoundException> indicesThatCannotBeCreated;
     private final Executor executor;
     private final LongSupplier relativeTimeProvider;
     private final FailureStoreDocumentConverter failureStoreDocumentConverter;
@@ -107,7 +106,6 @@ final class BulkOperation extends ActionRunnable<BulkResponse> {
         BulkRequest bulkRequest,
         NodeClient client,
         AtomicArray<BulkItemResponse> responses,
-        Map<String, IndexNotFoundException> indicesThatCannotBeCreated,
         IndexNameExpressionResolver indexNameExpressionResolver,
         LongSupplier relativeTimeProvider,
         long startTimeNanos,
@@ -122,7 +120,6 @@ final class BulkOperation extends ActionRunnable<BulkResponse> {
             bulkRequest,
             client,
             responses,
-            indicesThatCannotBeCreated,
             indexNameExpressionResolver,
             relativeTimeProvider,
             startTimeNanos,
@@ -141,7 +138,6 @@ final class BulkOperation extends ActionRunnable<BulkResponse> {
         BulkRequest bulkRequest,
         NodeClient client,
         AtomicArray<BulkItemResponse> responses,
-        Map<String, IndexNotFoundException> indicesThatCannotBeCreated,
         IndexNameExpressionResolver indexNameExpressionResolver,
         LongSupplier relativeTimeProvider,
         long startTimeNanos,
@@ -158,7 +154,6 @@ final class BulkOperation extends ActionRunnable<BulkResponse> {
         this.bulkRequest = bulkRequest;
         this.listener = listener;
         this.startTimeNanos = startTimeNanos;
-        this.indicesThatCannotBeCreated = indicesThatCannotBeCreated;
         this.executor = executor;
         this.relativeTimeProvider = relativeTimeProvider;
         this.indexNameExpressionResolver = indexNameExpressionResolver;
@@ -298,9 +293,6 @@ private Map<ShardId, List<BulkItemRequest>> groupRequestsByShards(
             if (addFailureIfRequiresAliasAndAliasIsMissing(docWriteRequest, bulkItemRequest.id(), metadata)) {
                 continue;
             }
-            if (addFailureIfIndexCannotBeCreated(docWriteRequest, bulkItemRequest.id())) {
-                continue;
-            }
             if (addFailureIfRequiresDataStreamAndNoParentDataStream(docWriteRequest, bulkItemRequest.id(), metadata)) {
                 continue;
             }
@@ -759,18 +751,6 @@ private boolean addFailureIfIndexIsClosed(DocWriteRequest<?> request, Index conc
         return false;
     }
 
-    private boolean addFailureIfIndexCannotBeCreated(DocWriteRequest<?> request, int idx) {
-        IndexNotFoundException cannotCreate = indicesThatCannotBeCreated.get(request.index());
-        if (cannotCreate != null) {
-            var failureStoreStatus = isFailureStoreRequest(request)
-                ? IndexDocFailureStoreStatus.FAILED
-                : IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN;
-            addFailureAndDiscardRequest(request, idx, request.index(), cannotCreate, failureStoreStatus);
-            return true;
-        }
-        return false;
-    }
-
     private static boolean isFailureStoreRequest(DocWriteRequest<?> request) {
         return request instanceof IndexRequest ir && ir.isWriteToFailureStore();
     }

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

@@ -43,7 +43,6 @@
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
 import org.elasticsearch.features.FeatureService;
-import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexingPressure;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.indices.SystemIndices;
@@ -60,6 +59,7 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.SortedMap;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 import java.util.function.Function;
 import java.util.function.LongSupplier;
@@ -351,29 +351,36 @@ protected void createMissingIndicesAndIndexData(
         final AtomicArray<BulkItemResponse> responses = new AtomicArray<>(bulkRequest.requests.size());
         // Optimizing when there are no prerequisite actions
         if (indicesToAutoCreate.isEmpty() && dataStreamsToBeRolledOver.isEmpty() && failureStoresToBeRolledOver.isEmpty()) {
-            executeBulk(task, bulkRequest, startTimeNanos, listener, executor, responses, Map.of());
+            executeBulk(task, bulkRequest, startTimeNanos, listener, executor, responses);
             return;
         }
-        final Map<String, IndexNotFoundException> indicesThatCannotBeCreated = new HashMap<>();
+        Map<String, Exception> indicesExceptions = new ConcurrentHashMap<>();
+        Map<String, Exception> dataStreamExceptions = new ConcurrentHashMap<>();
+        Map<String, Exception> failureStoreExceptions = new ConcurrentHashMap<>();
         Runnable executeBulkRunnable = () -> executor.execute(new ActionRunnable<>(listener) {
             @Override
             protected void doRun() {
-                executeBulk(task, bulkRequest, startTimeNanos, listener, executor, responses, indicesThatCannotBeCreated);
+                failRequestsWhenPrerequisiteActionFailed(
+                    indicesExceptions,
+                    dataStreamExceptions,
+                    failureStoreExceptions,
+                    bulkRequest,
+                    responses
+                );
+                executeBulk(task, bulkRequest, startTimeNanos, listener, executor, responses);
             }
         });
         try (RefCountingRunnable refs = new RefCountingRunnable(executeBulkRunnable)) {
-            createIndices(bulkRequest, indicesToAutoCreate, indicesThatCannotBeCreated, responses, refs);
-            rollOverDataStreams(bulkRequest, dataStreamsToBeRolledOver, false, responses, refs);
-            rollOverDataStreams(bulkRequest, failureStoresToBeRolledOver, true, responses, refs);
+            createIndices(indicesToAutoCreate, refs, indicesExceptions);
+            rollOverDataStreams(bulkRequest, dataStreamsToBeRolledOver, false, refs, dataStreamExceptions);
+            rollOverDataStreams(bulkRequest, failureStoresToBeRolledOver, true, refs, failureStoreExceptions);
         }
     }
 
     private void createIndices(
-        BulkRequest bulkRequest,
         Map<String, CreateIndexRequest> indicesToAutoCreate,
-        Map<String, IndexNotFoundException> indicesThatCannotBeCreated,
-        AtomicArray<BulkItemResponse> responses,
-        RefCountingRunnable refs
+        RefCountingRunnable refs,
+        final Map<String, Exception> indicesExceptions
     ) {
         for (Map.Entry<String, CreateIndexRequest> indexEntry : indicesToAutoCreate.entrySet()) {
             final String index = indexEntry.getKey();
@@ -384,25 +391,26 @@ public void onResponse(CreateIndexResponse createIndexResponse) {}
                 @Override
                 public void onFailure(Exception e) {
                     final Throwable cause = ExceptionsHelper.unwrapCause(e);
-                    if (cause instanceof IndexNotFoundException indexNotFoundException) {
-                        synchronized (indicesThatCannotBeCreated) {
-                            indicesThatCannotBeCreated.put(index, indexNotFoundException);
-                        }
-                    } else if ((cause instanceof ResourceAlreadyExistsException) == false) {
+                    if ((cause instanceof ResourceAlreadyExistsException) == false) {
                         // fail all requests involving this index, if create didn't work
-                        failRequestsWhenPrerequisiteActionFailed(index, bulkRequest, responses, e);
+                        indicesExceptions.put(index, e);
                     }
                 }
             }, refs.acquire()));
         }
     }
 
+    // Separate method to allow for overriding in tests.
+    void createIndex(CreateIndexRequest createIndexRequest, ActionListener<CreateIndexResponse> listener) {
+        client.execute(AutoCreateAction.INSTANCE, createIndexRequest, listener);
+    }
+
     private void rollOverDataStreams(
         BulkRequest bulkRequest,
         Set<String> dataStreamsToBeRolledOver,
         boolean targetFailureStore,
-        AtomicArray<BulkItemResponse> responses,
-        RefCountingRunnable refs
+        RefCountingRunnable refs,
+        Map<String, Exception> dataStreamExceptions
     ) {
         for (String dataStream : dataStreamsToBeRolledOver) {
             RolloverRequest rolloverRequest = new RolloverRequest(dataStream, null);
@@ -416,7 +424,7 @@ private void rollOverDataStreams(
             }
             // We are executing a lazy rollover because it is an action specialised for this situation, when we want an
             // unconditional and performant rollover.
-            rolloverClient.execute(LazyRolloverAction.INSTANCE, rolloverRequest, ActionListener.releaseAfter(new ActionListener<>() {
+            rollOver(rolloverRequest, ActionListener.releaseAfter(new ActionListener<>() {
 
                 @Override
                 public void onResponse(RolloverResponse result) {
@@ -431,26 +439,52 @@ public void onResponse(RolloverResponse result) {
 
                 @Override
                 public void onFailure(Exception e) {
-                    failRequestsWhenPrerequisiteActionFailed(dataStream, bulkRequest, responses, e);
+                    dataStreamExceptions.put(dataStream, e);
                 }
             }, refs.acquire()));
         }
     }
 
+    // Separate method to allow for overriding in tests.
+    void rollOver(RolloverRequest rolloverRequest, ActionListener<RolloverResponse> listener) {
+        rolloverClient.execute(LazyRolloverAction.INSTANCE, rolloverRequest, listener);
+    }
+
     /**
-     * Fails all requests involving this index or data stream because the prerequisite action failed too.
+     * Mark all the requests for which the prerequisite action failed (i.e. index creation or data stream/failure store rollover) as failed.
      */
-    private static void failRequestsWhenPrerequisiteActionFailed(
-        String target,
+    private void failRequestsWhenPrerequisiteActionFailed(
+        Map<String, Exception> indicesExceptions,
+        Map<String, Exception> dataStreamExceptions,
+        Map<String, Exception> failureStoreExceptions,
         BulkRequest bulkRequest,
-        AtomicArray<BulkItemResponse> responses,
-        Exception error
+        AtomicArray<BulkItemResponse> responses
     ) {
+        if (indicesExceptions.isEmpty() && dataStreamExceptions.isEmpty() && failureStoreExceptions.isEmpty()) {
+            return;
+        }
         for (int i = 0; i < bulkRequest.requests.size(); i++) {
             DocWriteRequest<?> request = bulkRequest.requests.get(i);
-            if (request != null && setResponseFailureIfIndexMatches(responses, i, request, target, error)) {
-                bulkRequest.requests.set(i, null);
+            if (request == null) {
+                continue;
             }
+            var exception = indicesExceptions.get(request.index());
+            if (exception == null) {
+                if (request instanceof IndexRequest indexRequest && indexRequest.isWriteToFailureStore()) {
+                    exception = failureStoreExceptions.get(request.index());
+                } else {
+                    exception = dataStreamExceptions.get(request.index());
+                }
+            }
+            if (exception == null) {
+                continue;
+            }
+            var failureStoreStatus = request instanceof IndexRequest ir && ir.isWriteToFailureStore()
+                ? IndexDocFailureStoreStatus.FAILED
+                : IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN;
+            var failure = new BulkItemResponse.Failure(request.index(), request.id(), exception, failureStoreStatus);
+            responses.set(i, BulkItemResponse.failure(i, request.opType(), failure));
+            bulkRequest.requests.set(i, null);
         }
     }
 
@@ -532,33 +566,13 @@ private static boolean isSystemIndex(SortedMap<String, IndexAbstraction> indices
         }
     }
 
-    void createIndex(CreateIndexRequest createIndexRequest, ActionListener<CreateIndexResponse> listener) {
-        client.execute(AutoCreateAction.INSTANCE, createIndexRequest, listener);
-    }
-
-    private static boolean setResponseFailureIfIndexMatches(
-        AtomicArray<BulkItemResponse> responses,
-        int idx,
-        DocWriteRequest<?> request,
-        String index,
-        Exception e
-    ) {
-        if (index.equals(request.index())) {
-            BulkItemResponse.Failure failure = new BulkItemResponse.Failure(request.index(), request.id(), e);
-            responses.set(idx, BulkItemResponse.failure(idx, request.opType(), failure));
-            return true;
-        }
-        return false;
-    }
-
     void executeBulk(
         Task task,
         BulkRequest bulkRequest,
         long startTimeNanos,
         ActionListener<BulkResponse> listener,
         Executor executor,
-        AtomicArray<BulkItemResponse> responses,
-        Map<String, IndexNotFoundException> indicesThatCannotBeCreated
+        AtomicArray<BulkItemResponse> responses
     ) {
         new BulkOperation(
             task,
@@ -568,7 +582,6 @@ void executeBulk(
             bulkRequest,
             client,
             responses,
-            indicesThatCannotBeCreated,
             indexNameExpressionResolver,
             relativeTimeNanosProvider,
             startTimeNanos,

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

@@ -42,7 +42,6 @@
 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.IndexVersion;
 import org.elasticsearch.index.mapper.MapperException;
 import org.elasticsearch.index.shard.ShardId;
@@ -1022,7 +1021,6 @@ private BulkOperation newBulkOperation(NodeClient client, BulkRequest request, A
             client,
             request,
             new AtomicArray<>(request.numberOfActions()),
-            Map.of(),
             mockObserver(DEFAULT_STATE),
             listener,
             new FailureStoreDocumentConverter()
@@ -1040,7 +1038,6 @@ private BulkOperation newBulkOperation(
             client,
             request,
             new AtomicArray<>(request.numberOfActions()),
-            Map.of(),
             mockObserver(DEFAULT_STATE),
             listener,
             failureStoreDocumentConverter
@@ -1059,7 +1056,6 @@ private BulkOperation newBulkOperation(
             client,
             request,
             new AtomicArray<>(request.numberOfActions()),
-            Map.of(),
             observer,
             listener,
             new FailureStoreDocumentConverter()
@@ -1071,7 +1067,6 @@ private BulkOperation newBulkOperation(
         NodeClient client,
         BulkRequest request,
         AtomicArray<BulkItemResponse> existingResponses,
-        Map<String, IndexNotFoundException> indicesThatCanNotBeCreated,
         ClusterStateObserver observer,
         ActionListener<BulkResponse> listener,
         FailureStoreDocumentConverter failureStoreDocumentConverter
@@ -1100,7 +1095,6 @@ private BulkOperation newBulkOperation(
             request,
             client,
             existingResponses,
-            indicesThatCanNotBeCreated,
             indexNameExpressionResolver,
             () -> endTime,
             timeZero,