Allow dropping documents with auto-generated ID (#46773)

When using auto-generated IDs + the ingest drop processor (which looks to be used by filebeat
as well) + coordinating nodes that do not have the ingest processor functionality, this can lead
to a NullPointerException.

The issue is that markCurrentItemAsDropped() is creating an UpdateResponse with no id when
the request contains auto-generated IDs. The response serialization is lenient for our
REST/XContent format (i.e. we will send "id" : null) but the internal transport format (used for
communication between nodes) assumes for this field to be non-null, which means that it can't
be serialized between nodes. Bulk requests with ingest functionality are processed on the
coordinating node if the node has the ingest capability, and only otherwise sent to a different
node. This means that, in order to reproduce this, one needs two nodes, with the coordinating
node not having the ingest functionality.

Closes #46678

Author: ywelsch

modules/reindex/src/test/java/org/elasticsearch/index/reindex/AsyncBulkByScrollActionTests.java

@@ -822,7 +822,7 @@ RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>>
                             new IndexResponse(
                                 shardId,
                                 index.type(),
-                                index.id(),
+                                index.id() == null ? "dummy_id" : index.id(),
                                 randomInt(20),
                                 randomIntBetween(1, 16),
                                 randomIntBetween(0, Integer.MAX_VALUE),

server/src/main/java/org/elasticsearch/action/DocWriteResponse.java

@@ -41,6 +41,7 @@
 import java.io.UnsupportedEncodingException;
 import java.net.URLEncoder;
 import java.util.Locale;
+import java.util.Objects;
 
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
 import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_PRIMARY_TERM;
@@ -122,13 +123,13 @@ public void writeTo(StreamOutput out) throws IOException {
     protected Result result;
 
     public DocWriteResponse(ShardId shardId, String type, String id, long seqNo, long primaryTerm, long version, Result result) {
-        this.shardId = shardId;
-        this.type = type;
-        this.id = id;
+        this.shardId = Objects.requireNonNull(shardId);
+        this.type = Objects.requireNonNull(type);
+        this.id = Objects.requireNonNull(id);
         this.seqNo = seqNo;
         this.primaryTerm = primaryTerm;
         this.version = version;
-        this.result = result;
+        this.result = Objects.requireNonNull(result);
     }
 
     // needed for deserialization

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

@@ -98,6 +98,7 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
     private final TransportCreateIndexAction createIndexAction;
     private final LongSupplier relativeTimeProvider;
     private final IngestActionForwarder ingestForwarder;
+    private static final String DROPPED_ITEM_WITH_AUTO_GENERATED_ID = "auto-generated";
 
     @Inject
     public TransportBulkAction(Settings settings, ThreadPool threadPool, TransportService transportService,
@@ -229,8 +230,8 @@ protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<Bulk
             final Set<String> indices = bulkRequest.requests.stream()
                     // delete requests should not attempt to create the index (if the index does not
                     // exists), unless an external versioning is used
-                .filter(request -> request.opType() != DocWriteRequest.OpType.DELETE 
-                        || request.versionType() == VersionType.EXTERNAL 
+                .filter(request -> request.opType() != DocWriteRequest.OpType.DELETE
+                        || request.versionType() == VersionType.EXTERNAL
                         || request.versionType() == VersionType.EXTERNAL_GTE)
                 .map(DocWriteRequest::index)
                 .collect(Collectors.toSet());
@@ -673,11 +674,12 @@ ActionListener<BulkResponse> wrapActionListenerIfNeeded(long ingestTookInMillis,
         void markCurrentItemAsDropped() {
             IndexRequest indexRequest = getIndexWriteRequest(bulkRequest.requests().get(currentSlot));
             failedSlots.set(currentSlot);
+            final String id = indexRequest.id() == null ? DROPPED_ITEM_WITH_AUTO_GENERATED_ID : indexRequest.id();
             itemResponses.add(
                 new BulkItemResponse(currentSlot, indexRequest.opType(),
                     new UpdateResponse(
                         new ShardId(indexRequest.index(), IndexMetaData.INDEX_UUID_NA_VALUE, 0),
-                        indexRequest.type(), indexRequest.id(), indexRequest.version(), DocWriteResponse.Result.NOOP
+                        indexRequest.type(), id, indexRequest.version(), DocWriteResponse.Result.NOOP
                     )
                 )
             );

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

@@ -26,6 +26,7 @@
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
+import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.client.NoOpClient;
 import org.junit.After;
@@ -226,7 +227,8 @@ public void bulk(BulkRequest request, ActionListener<BulkResponse> listener) {
         }
 
         private BulkItemResponse successfulResponse() {
-            return new BulkItemResponse(1, OpType.DELETE, new DeleteResponse());
+            return new BulkItemResponse(1, OpType.DELETE, new DeleteResponse(
+                new ShardId("test", "test", 0), "_doc", "test", 0, 0, 0, false));
         }
 
         private BulkItemResponse failedResponse() {

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

@@ -275,6 +275,7 @@ public void testExecuteBulkIndexRequestWithMappingUpdates() throws Exception {
         Engine.IndexResult success = new FakeIndexResult(1, 1, 13, true, resultLocation);
 
         IndexShard shard = mock(IndexShard.class);
+        when(shard.shardId()).thenReturn(shardId);
         when(shard.applyIndexOperationOnPrimary(anyLong(), any(), any(), anyLong(), anyLong(), anyLong(), anyBoolean()))
             .thenReturn(mappingUpdate);
 
@@ -606,6 +607,7 @@ public void testUpdateRequestWithSuccess() throws Exception {
         when(shard.applyIndexOperationOnPrimary(anyLong(), any(), any(), anyLong(), anyLong(), anyLong(), anyBoolean()))
             .thenReturn(indexResult);
         when(shard.indexSettings()).thenReturn(indexSettings);
+        when(shard.shardId()).thenReturn(shardId);
 
         UpdateHelper updateHelper = mock(UpdateHelper.class);
         when(updateHelper.prepare(any(), eq(shard), eq(true), any())).thenReturn(
@@ -652,6 +654,7 @@ public void testUpdateWithDelete() throws Exception {
         IndexShard shard = mock(IndexShard.class);
         when(shard.applyDeleteOperationOnPrimary(anyLong(), any(), any(), any(), anyLong(), anyLong())).thenReturn(deleteResult);
         when(shard.indexSettings()).thenReturn(indexSettings);
+        when(shard.shardId()).thenReturn(shardId);
 
         UpdateHelper updateHelper = mock(UpdateHelper.class);
         when(updateHelper.prepare(any(), eq(shard), eq(true), any())).thenReturn(
@@ -806,6 +809,7 @@ public void testRetries() throws Exception {
             }
         });
         when(shard.indexSettings()).thenReturn(indexSettings);
+        when(shard.shardId()).thenReturn(shardId);
 
         UpdateHelper updateHelper = mock(UpdateHelper.class);
         when(updateHelper.prepare(any(), eq(shard), eq(true), any())).thenReturn(
@@ -891,7 +895,7 @@ public void testRejectCASUsingSeqNo() throws Exception {
     private void randomlySetIgnoredPrimaryResponse(BulkItemRequest primaryRequest) {
         if (randomBoolean()) {
             // add a response to the request and thereby check that it is ignored for the primary.
-            primaryRequest.setPrimaryResponse(new BulkItemResponse(0, DocWriteRequest.OpType.INDEX, new IndexResponse(null, "_doc",
+            primaryRequest.setPrimaryResponse(new BulkItemResponse(0, DocWriteRequest.OpType.INDEX, new IndexResponse(shardId, "_doc",
                 "ignore-primary-response-on-primary", 42, 42, 42, false)));
         }
     }

server/src/test/java/org/elasticsearch/action/ingest/SimulateExecutionServiceTests.java

@@ -103,8 +103,7 @@ public void testExecuteItem() throws Exception {
 
     public void testExecuteVerboseItemExceptionWithoutOnFailure() throws Exception {
         TestProcessor processor1 = new TestProcessor("processor_0", "mock", ingestDocument -> {});
-        TestProcessor processor2 = new TestProcessor("processor_1", "mock",
-                ingestDocument -> { throw new RuntimeException("processor failed"); });
+        TestProcessor processor2 = new TestProcessor("processor_1", "mock", new RuntimeException("processor failed"));
         TestProcessor processor3 = new TestProcessor("processor_2", "mock", ingestDocument -> {});
         Pipeline pipeline = new Pipeline("_id", "_description", version, new CompoundProcessor(processor1, processor2, processor3));
         SimulateDocumentResult actualItemResponse = executionService.executeDocument(pipeline, ingestDocument, true);
@@ -128,8 +127,7 @@ public void testExecuteVerboseItemExceptionWithoutOnFailure() throws Exception {
     }
 
     public void testExecuteVerboseItemWithOnFailure() throws Exception {
-        TestProcessor processor1 = new TestProcessor("processor_0", "mock",
-                ingestDocument -> { throw new RuntimeException("processor failed"); });
+        TestProcessor processor1 = new TestProcessor("processor_0", "mock", new RuntimeException("processor failed"));
         TestProcessor processor2 = new TestProcessor("processor_1", "mock", ingestDocument -> {});
         TestProcessor processor3 = new TestProcessor("processor_2", "mock", ingestDocument -> {});
         Pipeline pipeline = new Pipeline("_id", "_description", version,
@@ -167,7 +165,7 @@ public void testExecuteVerboseItemWithOnFailure() throws Exception {
 
     public void testExecuteVerboseItemExceptionWithIgnoreFailure() throws Exception {
         RuntimeException exception = new RuntimeException("processor failed");
-        TestProcessor testProcessor = new TestProcessor("processor_0", "mock", ingestDocument -> { throw exception; });
+        TestProcessor testProcessor = new TestProcessor("processor_0", "mock", exception);
         CompoundProcessor processor = new CompoundProcessor(true, Collections.singletonList(testProcessor), Collections.emptyList());
         Pipeline pipeline = new Pipeline("_id", "_description", version, new CompoundProcessor(processor));
         SimulateDocumentResult actualItemResponse = executionService.executeDocument(pipeline, ingestDocument, true);

server/src/test/java/org/elasticsearch/index/mapper/DateFieldMapperTests.java

@@ -96,6 +96,23 @@ public void testDefaults() throws Exception {
         assertFalse(dvField.fieldType().stored());
     }
 
+    public void testSimple() throws Exception {
+        String mapping = Strings.toString(XContentFactory.jsonBuilder().startObject().startObject("type")
+            .startObject("properties").startObject("field").field("type", "date").endObject().endObject()
+            .endObject().endObject());
+
+        DocumentMapper mapper = parser.parse("type", new CompressedXContent(mapping));
+
+        assertEquals(mapping, mapper.mappingSource().toString());
+
+        ParsedDocument doc = mapper.parse(SourceToParse.source("test", "type", "1", BytesReference
+            .bytes(XContentFactory.jsonBuilder()
+                .startObject()
+                .field("field", "2016-03")
+                .endObject()),
+            XContentType.JSON));
+    }
+
     public void testNotIndexed() throws Exception {
         String mapping = Strings.toString(XContentFactory.jsonBuilder().startObject().startObject("type")
                 .startObject("properties").startObject("field").field("type", "date").field("index", false).endObject().endObject()

server/src/test/java/org/elasticsearch/ingest/CompoundProcessorTests.java

@@ -28,6 +28,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 import java.util.function.LongSupplier;
 
 import static org.hamcrest.CoreMatchers.equalTo;
@@ -74,7 +75,7 @@ public void testSingleProcessor() throws Exception {
     }
 
     public void testSingleProcessorWithException() throws Exception {
-        TestProcessor processor = new TestProcessor(ingestDocument -> {throw new RuntimeException("error");});
+        TestProcessor processor = new TestProcessor(new RuntimeException("error"));
         LongSupplier relativeTimeProvider = mock(LongSupplier.class);
         when(relativeTimeProvider.getAsLong()).thenReturn(0L);
         CompoundProcessor compoundProcessor = new CompoundProcessor(relativeTimeProvider, processor);
@@ -93,7 +94,7 @@ public void testSingleProcessorWithException() throws Exception {
     }
 
     public void testIgnoreFailure() throws Exception {
-        TestProcessor processor1 = new TestProcessor(ingestDocument -> {throw new RuntimeException("error");});
+        TestProcessor processor1 = new TestProcessor(new RuntimeException("error"));
         TestProcessor processor2 = new TestProcessor(ingestDocument -> {ingestDocument.setFieldValue("field", "value");});
         LongSupplier relativeTimeProvider = mock(LongSupplier.class);
         when(relativeTimeProvider.getAsLong()).thenReturn(0L);
@@ -108,7 +109,7 @@ public void testIgnoreFailure() throws Exception {
     }
 
     public void testSingleProcessorWithOnFailureProcessor() throws Exception {
-        TestProcessor processor1 = new TestProcessor("id", "first", ingestDocument -> {throw new RuntimeException("error");});
+        TestProcessor processor1 = new TestProcessor("id", "first", new RuntimeException("error"));
         TestProcessor processor2 = new TestProcessor(ingestDocument -> {
             Map<String, Object> ingestMetadata = ingestDocument.getIngestMetadata();
             assertThat(ingestMetadata.size(), equalTo(3));
@@ -130,7 +131,7 @@ public void testSingleProcessorWithOnFailureProcessor() throws Exception {
     }
 
     public void testSingleProcessorWithOnFailureDropProcessor() throws Exception {
-        TestProcessor processor1 = new TestProcessor("id", "first", ingestDocument -> {throw new RuntimeException("error");});
+        TestProcessor processor1 = new TestProcessor("id", "first", new RuntimeException("error"));
         Processor processor2 = new Processor() {
             @Override
             public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
@@ -159,8 +160,8 @@ public String getTag() {
     }
 
     public void testSingleProcessorWithNestedFailures() throws Exception {
-        TestProcessor processor = new TestProcessor("id", "first", ingestDocument -> {throw new RuntimeException("error");});
-        TestProcessor processorToFail = new TestProcessor("id2", "second", ingestDocument -> {
+        TestProcessor processor = new TestProcessor("id", "first", new RuntimeException("error"));
+        TestProcessor processorToFail = new TestProcessor("id2", "second", (Consumer<IngestDocument>) ingestDocument -> {
             Map<String, Object> ingestMetadata = ingestDocument.getIngestMetadata();
             assertThat(ingestMetadata.size(), equalTo(3));
             assertThat(ingestMetadata.get(CompoundProcessor.ON_FAILURE_MESSAGE_FIELD), equalTo("error"));
@@ -189,7 +190,7 @@ public void testSingleProcessorWithNestedFailures() throws Exception {
     }
 
     public void testCompoundProcessorExceptionFailWithoutOnFailure() throws Exception {
-        TestProcessor firstProcessor = new TestProcessor("id1", "first", ingestDocument -> {throw new RuntimeException("error");});
+        TestProcessor firstProcessor = new TestProcessor("id1", "first", new RuntimeException("error"));
         TestProcessor secondProcessor = new TestProcessor("id3", "second", ingestDocument -> {
             Map<String, Object> ingestMetadata = ingestDocument.getIngestMetadata();
             assertThat(ingestMetadata.entrySet(), hasSize(3));
@@ -212,9 +213,9 @@ public void testCompoundProcessorExceptionFailWithoutOnFailure() throws Exceptio
     }
 
     public void testCompoundProcessorExceptionFail() throws Exception {
-        TestProcessor firstProcessor = new TestProcessor("id1", "first", ingestDocument -> {throw new RuntimeException("error");});
+        TestProcessor firstProcessor = new TestProcessor("id1", "first", new RuntimeException("error"));
         TestProcessor failProcessor =
-            new TestProcessor("tag_fail", "fail", ingestDocument -> {throw new RuntimeException("custom error message");});
+            new TestProcessor("tag_fail", "fail", new RuntimeException("custom error message"));
         TestProcessor secondProcessor = new TestProcessor("id3", "second", ingestDocument -> {
             Map<String, Object> ingestMetadata = ingestDocument.getIngestMetadata();
             assertThat(ingestMetadata.entrySet(), hasSize(3));
@@ -238,9 +239,9 @@ public void testCompoundProcessorExceptionFail() throws Exception {
     }
 
     public void testCompoundProcessorExceptionFailInOnFailure() throws Exception {
-        TestProcessor firstProcessor = new TestProcessor("id1", "first", ingestDocument -> {throw new RuntimeException("error");});
+        TestProcessor firstProcessor = new TestProcessor("id1", "first", new RuntimeException("error"));
         TestProcessor failProcessor =
-            new TestProcessor("tag_fail", "fail", ingestDocument -> {throw new RuntimeException("custom error message");});
+            new TestProcessor("tag_fail", "fail", new RuntimeException("custom error message"));
         TestProcessor secondProcessor = new TestProcessor("id3", "second", ingestDocument -> {
             Map<String, Object> ingestMetadata = ingestDocument.getIngestMetadata();
             assertThat(ingestMetadata.entrySet(), hasSize(3));
@@ -264,8 +265,8 @@ public void testCompoundProcessorExceptionFailInOnFailure() throws Exception {
     }
 
     public void testBreakOnFailure() throws Exception {
-        TestProcessor firstProcessor = new TestProcessor("id1", "first", ingestDocument -> {throw new RuntimeException("error1");});
-        TestProcessor secondProcessor = new TestProcessor("id2", "second", ingestDocument -> {throw new RuntimeException("error2");});
+        TestProcessor firstProcessor = new TestProcessor("id1", "first", new RuntimeException("error1"));
+        TestProcessor secondProcessor = new TestProcessor("id2", "second", new RuntimeException("error2"));
         TestProcessor onFailureProcessor = new TestProcessor("id2", "on_failure", ingestDocument -> {});
         LongSupplier relativeTimeProvider = mock(LongSupplier.class);
         when(relativeTimeProvider.getAsLong()).thenReturn(0L);

server/src/test/java/org/elasticsearch/ingest/IngestClientIT.java

@@ -273,4 +273,25 @@ public void testPutWithPipelineFactoryError() throws Exception {
         GetPipelineResponse response = client().admin().cluster().prepareGetPipeline("_id2").get();
         assertFalse(response.isFound());
     }
+
+    public void testWithDedicatedMaster() throws Exception {
+        String masterOnlyNode = internalCluster().startMasterOnlyNode();
+        BytesReference source = BytesReference.bytes(jsonBuilder().startObject()
+            .field("description", "my_pipeline")
+            .startArray("processors")
+            .startObject()
+            .startObject("test")
+            .endObject()
+            .endObject()
+            .endArray()
+            .endObject());
+        PutPipelineRequest putPipelineRequest = new PutPipelineRequest("_id", source, XContentType.JSON);
+        client().admin().cluster().putPipeline(putPipelineRequest).get();
+
+        BulkItemResponse item = client(masterOnlyNode).prepareBulk().add(
+            client().prepareIndex("test", "type").setSource("field", "value2", "drop", true).setPipeline("_id")).get()
+            .getItems()[0];
+        assertFalse(item.isFailed());
+        assertEquals("auto-generated", item.getResponse().getId());
+    }
 }

server/src/test/java/org/elasticsearch/ingest/TrackingResultProcessorTests.java

@@ -101,7 +101,7 @@ public void testActualCompoundProcessorWithoutOnFailure() throws Exception {
 
     public void testActualCompoundProcessorWithOnFailure() throws Exception {
         RuntimeException exception = new RuntimeException("fail");
-        TestProcessor failProcessor = new TestProcessor("fail", "test", ingestDocument -> { throw exception; });
+        TestProcessor failProcessor = new TestProcessor("fail", "test", exception);
         TestProcessor onFailureProcessor = new TestProcessor("success", "test", ingestDocument -> {});
         CompoundProcessor actualProcessor = new CompoundProcessor(false,
             Arrays.asList(new CompoundProcessor(false,

test/framework/src/main/java/org/elasticsearch/ingest/IngestTestPlugin.java

@@ -37,6 +37,10 @@ public Map<String, Processor.Factory> getProcessors(Processor.Parameters paramet
                 if (doc.hasField("fail") && doc.getFieldValue("fail", Boolean.class)) {
                     throw new IllegalArgumentException("test processor failed");
                 }
+                if (doc.hasField("drop") && doc.getFieldValue("drop", Boolean.class)) {
+                    return null;
+                }
+                return doc;
             }));
     }
 }