Allow dropping documents with auto-generated ID

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

@@ -40,6 +40,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;
@@ -121,13 +122,13 @@ public void writeTo(StreamOutput out) throws IOException {
     protected final 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

@@ -672,11 +672,12 @@ ActionListener<BulkResponse> wrapActionListenerIfNeeded(long ingestTookInMillis,
         void markCurrentItemAsDropped() {
             IndexRequest indexRequest = getIndexWriteRequest(bulkRequest.requests().get(currentSlot));
             failedSlots.set(currentSlot);
+            final String id = indexRequest.id() == null ? "auto-generated" : 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(), SequenceNumbers.UNASSIGNED_SEQ_NO, SequenceNumbers.UNASSIGNED_PRIMARY_TERM,
+                        indexRequest.type(), id, SequenceNumbers.UNASSIGNED_SEQ_NO, SequenceNumbers.UNASSIGNED_PRIMARY_TERM,
                         indexRequest.version(), DocWriteResponse.Result.NOOP
                     )
                 )

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

@@ -272,4 +272,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());
+    }
 }

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;
             }));
     }
 }

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

@@ -22,6 +22,7 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 /**
  * Processor used for testing, keeps track of how many times it is invoked and
@@ -31,24 +32,30 @@ public class TestProcessor implements Processor {
 
     private final String type;
     private final String tag;
-    private final Consumer<IngestDocument> ingestDocumentConsumer;
+    private final Function<IngestDocument, IngestDocument> ingestDocumentMapper;
     private final AtomicInteger invokedCounter = new AtomicInteger();
 
     public TestProcessor(Consumer<IngestDocument> ingestDocumentConsumer) {
         this(null, "test-processor", ingestDocumentConsumer);
     }
 
     public TestProcessor(String tag, String type, Consumer<IngestDocument> ingestDocumentConsumer) {
-        this.ingestDocumentConsumer = ingestDocumentConsumer;
+        this(tag, type, id -> {
+            ingestDocumentConsumer.accept(id);
+            return id;
+        });
+    }
+
+    public TestProcessor(String tag, String type, Function<IngestDocument, IngestDocument> ingestDocumentMapper) {
+        this.ingestDocumentMapper = ingestDocumentMapper;
         this.type = type;
         this.tag = tag;
     }
 
     @Override
     public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
         invokedCounter.incrementAndGet();
-        ingestDocumentConsumer.accept(ingestDocument);
-        return ingestDocument;
+        return ingestDocumentMapper.apply(ingestDocument);
     }
 
     @Override

test/framework/src/main/java/org/elasticsearch/test/InternalTestCluster.java

@@ -2002,6 +2002,7 @@ public String startMasterOnlyNode(Settings settings) {
                 .put(settings)
                 .put(Node.NODE_MASTER_SETTING.getKey(), true)
                 .put(Node.NODE_DATA_SETTING.getKey(), false)
+                .put(Node.NODE_INGEST_SETTING.getKey(), false)
                 .build();
         return startNode(settings1);
     }