Simplify bulk request execution

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/RestNoopBulkAction.java

@@ -18,6 +18,7 @@
  */
 package org.elasticsearch.plugin.noop.action.bulk;
 
+import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequest;
@@ -82,7 +83,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
     }
 
     private static class BulkRestBuilderListener extends RestBuilderListener<BulkRequest> {
-        private final BulkItemResponse ITEM_RESPONSE = new BulkItemResponse(1, "update",
+        private final BulkItemResponse ITEM_RESPONSE = new BulkItemResponse(1, DocWriteRequest.OpType.UPDATE,
             new UpdateResponse(new ShardId("mock", "", 1), "mock_type", "1", 1L, DocWriteResponse.Result.CREATED));
 
         private final RestRequest request;

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/TransportNoopBulkAction.java

@@ -19,6 +19,7 @@
 package org.elasticsearch.plugin.noop.action.bulk;
 
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequest;
@@ -34,7 +35,7 @@
 import org.elasticsearch.transport.TransportService;
 
 public class TransportNoopBulkAction extends HandledTransportAction<BulkRequest, BulkResponse> {
-    private static final BulkItemResponse ITEM_RESPONSE = new BulkItemResponse(1, "update",
+    private static final BulkItemResponse ITEM_RESPONSE = new BulkItemResponse(1, DocWriteRequest.OpType.UPDATE,
         new UpdateResponse(new ShardId("mock", "", 1), "mock_type", "1", 1L, DocWriteResponse.Result.CREATED));
 
     @Inject

core/src/main/java/org/elasticsearch/action/DocWriteRequest.java

@@ -0,0 +1,203 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.action;
+
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.index.VersionType;
+
+import java.io.IOException;
+import java.util.Locale;
+
+/**
+ * Generic interface to group ActionRequest, which perform writes to a single document
+ * Action requests implementing this can be part of {@link org.elasticsearch.action.bulk.BulkRequest}
+ */
+public interface DocWriteRequest<T> extends IndicesRequest {
+
+    /**
+     * Get the index that this request operates on
+     * @return the index
+     */
+    String index();
+
+    /**
+     * Get the type that this request operates on
+     * @return the type
+     */
+    String type();
+
+    /**
+     * Get the id of the document for this request
+     * @return the id
+     */
+    String id();
+
+    /**
+     * Get the options for this request
+     * @return the indices options
+     */
+    IndicesOptions indicesOptions();
+
+    /**
+     * Set the routing for this request
+     * @return the Request
+     */
+    T routing(String routing);
+
+    /**
+     * Get the routing for this request
+     * @return the Routing
+     */
+    String routing();
+
+
+    /**
+     * Get the parent for this request
+     * @return the Parent
+     */
+    String parent();
+
+    /**
+     * Get the document version for this request
+     * @return the document version
+     */
+    long version();
+
+    /**
+     * Sets the version, which will perform the operation only if a matching
+     * version exists and no changes happened on the doc since then.
+     */
+    T version(long version);
+
+    /**
+     * Get the document version type for this request
+     * @return the document version type
+     */
+    VersionType versionType();
+
+    /**
+     * Sets the versioning type. Defaults to {@link VersionType#INTERNAL}.
+     */
+    T versionType(VersionType versionType);
+
+    /**
+     * Get the requested document operation type of the request
+     * @return the operation type {@link OpType}
+     */
+    OpType opType();
+
+    /**
+     * Requested operation type to perform on the document
+     */
+    enum OpType {
+        /**
+         * Index the source. If there an existing document with the id, it will
+         * be replaced.
+         */
+        INDEX(0),
+        /**
+         * Creates the resource. Simply adds it to the index, if there is an existing
+         * document with the id, then it won't be removed.
+         */
+        CREATE(1),
+        /** Updates a document */
+        UPDATE(2),
+        /** Deletes a document */
+        DELETE(3);
+
+        private final byte op;
+        private final String lowercase;
+
+        OpType(int op) {
+            this.op = (byte) op;
+            this.lowercase = this.toString().toLowerCase(Locale.ROOT);
+        }
+
+        public byte getId() {
+            return op;
+        }
+
+        public String getLowercase() {
+            return lowercase;
+        }
+
+        public static OpType fromId(byte id) {
+            switch (id) {
+                case 0: return INDEX;
+                case 1: return CREATE;
+                case 2: return UPDATE;
+                case 3: return DELETE;
+                default: throw new IllegalArgumentException("Unknown opType: [" + id + "]");
+            }
+        }
+
+        public static OpType fromString(String sOpType) {
+            String lowerCase = sOpType.toLowerCase(Locale.ROOT);
+            for (OpType opType : OpType.values()) {
+                if (opType.getLowercase().equals(lowerCase)) {
+                    return opType;
+                }
+            }
+            throw new IllegalArgumentException("Unknown opType: [" + sOpType + "]");
+        }
+    }
+
+    /** read a document write (index/delete/update) request */
+    static DocWriteRequest readDocumentRequest(StreamInput in) throws IOException {
+        byte type = in.readByte();
+        DocWriteRequest docWriteRequest;
+        if (type == 0) {
+            IndexRequest indexRequest = new IndexRequest();
+            indexRequest.readFrom(in);
+            docWriteRequest = indexRequest;
+        } else if (type == 1) {
+            DeleteRequest deleteRequest = new DeleteRequest();
+            deleteRequest.readFrom(in);
+            docWriteRequest = deleteRequest;
+        } else if (type == 2) {
+            UpdateRequest updateRequest = new UpdateRequest();
+            updateRequest.readFrom(in);
+            docWriteRequest = updateRequest;
+        } else {
+            throw new IllegalStateException("invalid request type [" + type+ " ]");
+        }
+        return docWriteRequest;
+    }
+
+    /** write a document write (index/delete/update) request*/
+    static void writeDocumentRequest(StreamOutput out, DocWriteRequest request)  throws IOException {
+        if (request instanceof IndexRequest) {
+            out.writeByte((byte) 0);
+            ((IndexRequest) request).writeTo(out);
+        } else if (request instanceof DeleteRequest) {
+            out.writeByte((byte) 1);
+            ((DeleteRequest) request).writeTo(out);
+        } else if (request instanceof UpdateRequest) {
+            out.writeByte((byte) 2);
+            ((UpdateRequest) request).writeTo(out);
+        } else {
+            throw new IllegalStateException("invalid request [" + request.getClass().getSimpleName() + " ]");
+        }
+    }
+}

core/src/main/java/org/elasticsearch/action/bulk/BulkItemRequest.java

@@ -19,7 +19,7 @@
 
 package org.elasticsearch.action.bulk;
 
-import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.index.IndexRequest;
@@ -36,15 +36,15 @@
 public class BulkItemRequest implements Streamable {
 
     private int id;
-    private ActionRequest request;
+    private DocWriteRequest request;
     private volatile BulkItemResponse primaryResponse;
     private volatile boolean ignoreOnReplica;
 
     BulkItemRequest() {
 
     }
 
-    public BulkItemRequest(int id, ActionRequest request) {
+    public BulkItemRequest(int id, DocWriteRequest request) {
         assert request instanceof IndicesRequest;
         this.id = id;
         this.request = request;
@@ -54,14 +54,13 @@ public int id() {
         return id;
     }
 
-    public ActionRequest request() {
+    public DocWriteRequest request() {
         return request;
     }
 
     public String index() {
-        IndicesRequest indicesRequest = (IndicesRequest) request;
-        assert indicesRequest.indices().length == 1;
-        return indicesRequest.indices()[0];
+        assert request.indices().length == 1;
+        return request.indices()[0];
     }
 
     BulkItemResponse getPrimaryResponse() {
@@ -92,15 +91,7 @@ public static BulkItemRequest readBulkItem(StreamInput in) throws IOException {
     @Override
     public void readFrom(StreamInput in) throws IOException {
         id = in.readVInt();
-        byte type = in.readByte();
-        if (type == 0) {
-            request = new IndexRequest();
-        } else if (type == 1) {
-            request = new DeleteRequest();
-        } else if (type == 2) {
-            request = new UpdateRequest();
-        }
-        request.readFrom(in);
+        request = DocWriteRequest.readDocumentRequest(in);
         if (in.readBoolean()) {
             primaryResponse = BulkItemResponse.readBulkItem(in);
         }
@@ -110,14 +101,7 @@ public void readFrom(StreamInput in) throws IOException {
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeVInt(id);
-        if (request instanceof IndexRequest) {
-            out.writeByte((byte) 0);
-        } else if (request instanceof DeleteRequest) {
-            out.writeByte((byte) 1);
-        } else if (request instanceof UpdateRequest) {
-            out.writeByte((byte) 2);
-        }
-        request.writeTo(out);
+        DocWriteRequest.writeDocumentRequest(out, request);
         out.writeOptionalStreamable(primaryResponse);
         out.writeBoolean(ignoreOnReplica);
     }