Better handling of async processor failures (#104289) (#104768)

Author: joegallo

docs/changelog/104289.yaml

@@ -0,0 +1,6 @@
+pr: 104289
+summary: Better handling of async processor failures
+area: Ingest Node
+type: bug
+issues:
+ - 101921

server/src/internalClusterTest/java/org/elasticsearch/action/ingest/AsyncIngestProcessorIT.java

@@ -7,11 +7,14 @@
  */
 package org.elasticsearch.action.ingest;
 
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.get.GetRequest;
 import org.elasticsearch.action.get.GetResponse;
 import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.update.UpdateResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -40,18 +43,26 @@
 import java.util.function.BiConsumer;
 import java.util.function.Supplier;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
 /**
  * The purpose of this test is to verify that when a processor executes an operation asynchronously that
  * the expected result is the same as if the same operation happens synchronously.
- *
- * In this test two test processor are defined that basically do the same operation, but a single processor
+ * <p>
+ * In this test two test processors are defined that basically do the same operation, but a single processor
  * executes asynchronously. The result of the operation should be the same and also the order in which the
  * bulk responses are returned should be the same as how the corresponding index requests were defined.
+ * <p>
+ * As a further test, one document is dropped by the synchronous processor, and one document causes
+ * the asynchronous processor throw an exception.
  */
 public class AsyncIngestProcessorIT extends ESSingleNodeTestCase {
 
+    private static final int DROPPED = 3;
+
+    private static final int ERROR = 7;
+
     @Override
     protected Collection<Class<? extends Plugin>> getPlugins() {
         return Collections.singleton(TestPlugin.class);
@@ -72,11 +83,21 @@ public void testAsyncProcessorImplementation() {
         for (int i = 0; i < numDocs; i++) {
             String id = Integer.toString(i);
             assertThat(bulkResponse.getItems()[i].getId(), equalTo(id));
-            GetResponse getResponse = client().get(new GetRequest("foobar", id)).actionGet();
-            // The expected result of async test processor:
-            assertThat(getResponse.getSource().get("foo"), equalTo("bar-" + id));
-            // The expected result of sync test processor:
-            assertThat(getResponse.getSource().get("bar"), equalTo("baz-" + id));
+            if (i == DROPPED) {
+                UpdateResponse dropped = bulkResponse.getItems()[i].getResponse();
+                assertThat(dropped.getId(), equalTo(id));
+                assertThat(dropped.getResult(), equalTo(DocWriteResponse.Result.NOOP));
+            } else if (i == ERROR) {
+                BulkItemResponse failure = bulkResponse.getItems()[i];
+                assertThat(failure.getFailure().getId(), equalTo(id));
+                assertThat(failure.getFailure().getMessage(), containsString("lucky number seven"));
+            } else {
+                GetResponse getResponse = client().get(new GetRequest("foobar", id)).actionGet();
+                // The expected result of async test processor:
+                assertThat(getResponse.getSource().get("foo"), equalTo("bar-" + id));
+                // The expected result of sync test processor:
+                assertThat(getResponse.getSource().get("bar"), equalTo("baz-" + id));
+            }
         }
     }
 
@@ -112,15 +133,20 @@ public Map<String, Processor.Factory> getProcessors(Processor.Parameters paramet
                     public void execute(IngestDocument ingestDocument, BiConsumer<IngestDocument, Exception> handler) {
                         threadPool.generic().execute(() -> {
                             String id = (String) ingestDocument.getSourceAndMetadata().get("_id");
-                            if (usually()) {
-                                try {
-                                    Thread.sleep(10);
-                                } catch (InterruptedException e) {
-                                    // ignore
+                            if (id.equals(String.valueOf(ERROR))) {
+                                // lucky number seven always fails
+                                handler.accept(ingestDocument, new RuntimeException("lucky number seven"));
+                            } else {
+                                if (usually()) {
+                                    try {
+                                        Thread.sleep(10);
+                                    } catch (InterruptedException e) {
+                                        // ignore
+                                    }
                                 }
+                                ingestDocument.setFieldValue("foo", "bar-" + id);
+                                handler.accept(ingestDocument, null);
                             }
-                            ingestDocument.setFieldValue("foo", "bar-" + id);
-                            handler.accept(ingestDocument, null);
                         });
                     }
 
@@ -140,8 +166,13 @@ public String getType() {
                     @Override
                     public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
                         String id = (String) ingestDocument.getSourceAndMetadata().get("_id");
-                        ingestDocument.setFieldValue("bar", "baz-" + id);
-                        return ingestDocument;
+                        if (id.equals(String.valueOf(DROPPED))) {
+                            // lucky number three is always dropped
+                            return null;
+                        } else {
+                            ingestDocument.setFieldValue("bar", "baz-" + id);
+                            return ingestDocument;
+                        }
                     }
 
                     @Override

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

@@ -47,6 +47,7 @@
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
+import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.Index;
@@ -79,6 +80,7 @@
 import java.util.concurrent.atomic.AtomicIntegerArray;
 import java.util.function.LongSupplier;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import static org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.EXCLUDED_DATA_STREAMS_KEY;
 import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_PRIMARY_TERM;
@@ -893,19 +895,57 @@ ActionListener<BulkResponse> wrapActionListenerIfNeeded(long ingestTookInMillis,
                 );
             } else {
                 return actionListener.map(response -> {
-                    BulkItemResponse[] items = response.getItems();
-                    for (int i = 0; i < items.length; i++) {
-                        itemResponses.add(originalSlots.get(i), response.getItems()[i]);
+                    // these items are the responses from the subsequent bulk request, their 'slots'
+                    // are not correct for this response we're building
+                    final BulkItemResponse[] bulkResponses = response.getItems();
+
+                    final BulkItemResponse[] allResponses = new BulkItemResponse[bulkResponses.length + itemResponses.size()];
+
+                    // the item responses are from the original request, so their slots are correct.
+                    // these are the responses for requests that failed early and were not passed on to the subsequent bulk.
+                    for (BulkItemResponse item : itemResponses) {
+                        allResponses[item.getItemId()] = item;
                     }
-                    return new BulkResponse(
-                        itemResponses.toArray(new BulkItemResponse[0]),
-                        response.getTook().getMillis(),
-                        ingestTookInMillis
-                    );
+
+                    // use the original slots for the responses from the bulk
+                    for (int i = 0; i < bulkResponses.length; i++) {
+                        allResponses[originalSlots.get(i)] = bulkResponses[i];
+                    }
+
+                    if (Assertions.ENABLED) {
+                        assertResponsesAreCorrect(bulkResponses, allResponses);
+                    }
+
+                    return new BulkResponse(allResponses, response.getTook().getMillis(), ingestTookInMillis);
                 });
             }
         }
 
+        private void assertResponsesAreCorrect(BulkItemResponse[] bulkResponses, BulkItemResponse[] allResponses) {
+            // check for an empty intersection between the ids
+            final Set<Integer> failedIds = itemResponses.stream().map(BulkItemResponse::getItemId).collect(Collectors.toSet());
+            final Set<Integer> responseIds = IntStream.range(0, bulkResponses.length)
+                .map(originalSlots::get) // resolve subsequent bulk ids back to the original slots
+                .boxed()
+                .collect(Collectors.toSet());
+            assert Sets.haveEmptyIntersection(failedIds, responseIds)
+                : "bulk item response slots cannot have failed and been processed in the subsequent bulk request, failed ids: "
+                    + failedIds
+                    + ", response ids: "
+                    + responseIds;
+
+            // check for the correct number of responses
+            final int expectedResponseCount = bulkRequest.requests.size();
+            final int actualResponseCount = failedIds.size() + responseIds.size();
+            assert expectedResponseCount == actualResponseCount
+                : "Expected [" + expectedResponseCount + "] responses, but found [" + actualResponseCount + "]";
+
+            // check that every response is present
+            for (int i = 0; i < allResponses.length; i++) {
+                assert allResponses[i] != null : "BulkItemResponse at index [" + i + "] was null";
+            }
+        }
+
         synchronized void markItemAsDropped(int slot) {
             IndexRequest indexRequest = getIndexWriteRequest(bulkRequest.requests().get(slot));
             failedSlots.set(slot);
@@ -942,7 +982,7 @@ synchronized void markItemAsFailed(int slot, Exception e) {
             );
 
             // We hit a error during preprocessing a request, so we:
-            // 1) Remember the request item slot from the bulk, so that we're done processing all requests we know what failed
+            // 1) Remember the request item slot from the bulk, so that when we're done processing all requests we know what failed
             // 2) Add a bulk item failure for this request
             // 3) Continue with the next request in the bulk.
             failedSlots.set(slot);

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

@@ -15,18 +15,18 @@
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.XContentType;
-import org.hamcrest.Matchers;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.sameInstance;
 
 public class BulkRequestModifierTests extends ESTestCase {
 
@@ -36,38 +36,53 @@ public void testBulkRequestModifier() {
         for (int i = 0; i < numRequests; i++) {
             bulkRequest.add(new IndexRequest("_index", "_type", String.valueOf(i)).source("{}", XContentType.JSON));
         }
-        CaptureActionListener actionListener = new CaptureActionListener();
-        TransportBulkAction.BulkRequestModifier bulkRequestModifier = new TransportBulkAction.BulkRequestModifier(bulkRequest);
 
-        int i = 0;
+        // wrap the bulk request and fail some of the item requests at random
+        TransportBulkAction.BulkRequestModifier modifier = new TransportBulkAction.BulkRequestModifier(bulkRequest);
         Set<Integer> failedSlots = new HashSet<>();
-        while (bulkRequestModifier.hasNext()) {
-            bulkRequestModifier.next();
+        for (int i = 0; modifier.hasNext(); i++) {
+            modifier.next();
             if (randomBoolean()) {
-                bulkRequestModifier.markItemAsFailed(i, new RuntimeException());
+                modifier.markItemAsFailed(i, new RuntimeException());
                 failedSlots.add(i);
             }
-            i++;
+        }
+        assertThat(modifier.getBulkRequest().requests().size(), equalTo(numRequests - failedSlots.size()));
+
+        // populate the non-failed responses
+        BulkRequest subsequentBulkRequest = modifier.getBulkRequest();
+        assertThat(subsequentBulkRequest.requests().size(), equalTo(numRequests - failedSlots.size()));
+        List<BulkItemResponse> responses = new ArrayList<>();
+        for (int j = 0; j < subsequentBulkRequest.requests().size(); j++) {
+            IndexRequest indexRequest = (IndexRequest) subsequentBulkRequest.requests().get(j);
+            IndexResponse indexResponse = new IndexResponse(new ShardId("_index", "_na_", 0), "_type", indexRequest.id(), 1, 17, 1, true);
+            responses.add(BulkItemResponse.success(j, indexRequest.opType(), indexResponse));
         }
 
-        assertThat(bulkRequestModifier.getBulkRequest().requests().size(), equalTo(numRequests - failedSlots.size()));
-        // simulate that we actually executed the modified bulk request:
+        // simulate that we actually executed the modified bulk request
         long ingestTook = randomLong();
-        ActionListener<BulkResponse> result = bulkRequestModifier.wrapActionListenerIfNeeded(ingestTook, actionListener);
-        result.onResponse(new BulkResponse(new BulkItemResponse[numRequests - failedSlots.size()], 0));
+        CaptureActionListener actionListener = new CaptureActionListener();
+        ActionListener<BulkResponse> result = modifier.wrapActionListenerIfNeeded(ingestTook, actionListener);
+        result.onResponse(new BulkResponse(responses.toArray(new BulkItemResponse[0]), 0));
 
+        // check the results for successes and failures
         BulkResponse bulkResponse = actionListener.getResponse();
         assertThat(bulkResponse.getIngestTookInMillis(), equalTo(ingestTook));
-        for (int j = 0; j < bulkResponse.getItems().length; j++) {
-            if (failedSlots.contains(j)) {
-                BulkItemResponse item = bulkResponse.getItems()[j];
+        for (int i = 0; i < bulkResponse.getItems().length; i++) {
+            BulkItemResponse item = bulkResponse.getItems()[i];
+            if (failedSlots.contains(i)) {
                 assertThat(item.isFailed(), is(true));
-                assertThat(item.getFailure().getIndex(), equalTo("_index"));
-                assertThat(item.getFailure().getType(), equalTo("_type"));
-                assertThat(item.getFailure().getId(), equalTo(String.valueOf(j)));
-                assertThat(item.getFailure().getMessage(), equalTo("java.lang.RuntimeException"));
+                BulkItemResponse.Failure failure = item.getFailure();
+                assertThat(failure.getIndex(), equalTo("_index"));
+                assertThat(failure.getType(), equalTo("_type"));
+                assertThat(failure.getId(), equalTo(String.valueOf(i)));
+                assertThat(failure.getMessage(), equalTo("java.lang.RuntimeException"));
             } else {
-                assertThat(bulkResponse.getItems()[j], nullValue());
+                assertThat(item.isFailed(), is(false));
+                IndexResponse success = item.getResponse();
+                assertThat(success.getIndex(), equalTo("_index"));
+                assertThat(success.getType(), equalTo("_type"));
+                assertThat(success.getId(), equalTo(String.valueOf(i)));
             }
         }
     }
@@ -79,16 +94,29 @@ public void testPipelineFailures() {
         }
 
         TransportBulkAction.BulkRequestModifier modifier = new TransportBulkAction.BulkRequestModifier(originalBulkRequest);
+
+        final List<Integer> failures = new ArrayList<>();
+        // iterate the requests in order, recording that half of them should be failures
         for (int i = 0; modifier.hasNext(); i++) {
             modifier.next();
             if (i % 2 == 0) {
-                modifier.markItemAsFailed(i, new RuntimeException());
+                failures.add(i);
             }
         }
 
+        // with async processors, the failures can come back 'out of order' so sometimes we'll shuffle the list
+        if (randomBoolean()) {
+            Collections.shuffle(failures, random());
+        }
+
+        // actually mark the failures
+        for (int i : failures) {
+            modifier.markItemAsFailed(i, new RuntimeException());
+        }
+
         // So half of the requests have "failed", so only the successful requests are left:
         BulkRequest bulkRequest = modifier.getBulkRequest();
-        assertThat(bulkRequest.requests().size(), Matchers.equalTo(16));
+        assertThat(bulkRequest.requests().size(), equalTo(16));
 
         List<BulkItemResponse> responses = new ArrayList<>();
         ActionListener<BulkResponse> bulkResponseListener = modifier.wrapActionListenerIfNeeded(1L, new ActionListener<BulkResponse>() {
@@ -115,11 +143,11 @@ public void onFailure(Exception e) {}
             );
             originalResponses.add(BulkItemResponse.success(Integer.parseInt(indexRequest.id()), indexRequest.opType(), indexResponse));
         }
-        bulkResponseListener.onResponse(new BulkResponse(originalResponses.toArray(new BulkItemResponse[originalResponses.size()]), 0));
+        bulkResponseListener.onResponse(new BulkResponse(originalResponses.toArray(new BulkItemResponse[0]), 0));
 
-        assertThat(responses.size(), Matchers.equalTo(32));
+        assertThat(responses.size(), equalTo(32));
         for (int i = 0; i < 32; i++) {
-            assertThat(responses.get(i).getId(), Matchers.equalTo(String.valueOf(i)));
+            assertThat(responses.get(i).getId(), equalTo(String.valueOf(i)));
         }
     }
 
@@ -135,7 +163,7 @@ public void testNoFailures() {
         }
 
         BulkRequest bulkRequest = modifier.getBulkRequest();
-        assertThat(bulkRequest, Matchers.sameInstance(originalBulkRequest));
+        assertThat(bulkRequest, sameInstance(originalBulkRequest));
         ActionListener<BulkResponse> actionListener = ActionListener.wrap(() -> {});
         assertThat(modifier.wrapActionListenerIfNeeded(1L, actionListener), instanceOf(ActionListener.MappedActionListener.class));
     }