Index Request wrongly stores extra bytes as part of the document (trailing null bytes), closes #175.

Author: kimchy

modules/elasticsearch/src/main/java/org/elasticsearch/action/count/CountRequest.java

@@ -213,17 +213,17 @@ int querySourceLength() {
      * The query source to execute.
      */
     @Required public CountRequest query(byte[] querySource) {
-        return query(querySource, 0, querySource.length);
+        return query(querySource, 0, querySource.length, false);
     }
 
     /**
      * The query source to execute.
      */
-    @Required public CountRequest query(byte[] querySource, int offset, int length) {
+    @Required public CountRequest query(byte[] querySource, int offset, int length, boolean unsafe) {
         this.querySource = querySource;
         this.querySourceOffset = offset;
         this.querySourceLength = length;
-        this.querySourceUnsafe = false;
+        this.querySourceUnsafe = unsafe;
         return this;
     }
 

modules/elasticsearch/src/main/java/org/elasticsearch/action/deletebyquery/DeleteByQueryRequest.java

@@ -103,7 +103,7 @@ public DeleteByQueryRequest indices(String... indices) {
      * The query source to execute.
      */
     byte[] querySource() {
-        if (querySourceUnsafe) {
+        if (querySourceUnsafe || querySourceOffset > 0) {
             querySource = Arrays.copyOfRange(querySource, querySourceOffset, querySourceLength);
             querySourceOffset = 0;
             querySourceUnsafe = false;
@@ -167,17 +167,17 @@ byte[] querySource() {
      * The query source to execute.
      */
     @Required public DeleteByQueryRequest query(byte[] querySource) {
-        return query(querySource, 0, querySource.length);
+        return query(querySource, 0, querySource.length, false);
     }
 
     /**
      * The query source to execute.
      */
-    @Required public DeleteByQueryRequest query(byte[] querySource, int offset, int length) {
+    @Required public DeleteByQueryRequest query(byte[] querySource, int offset, int length, boolean unsafe) {
         this.querySource = querySource;
         this.querySourceOffset = offset;
         this.querySourceLength = length;
-        this.querySourceUnsafe = false;
+        this.querySourceUnsafe = unsafe;
         return this;
     }
 

modules/elasticsearch/src/main/java/org/elasticsearch/action/index/IndexRequest.java

@@ -126,10 +126,9 @@ public IndexRequest(String index) {
     /**
      * Constructs a new index request against the index, type, id and using the source.
      *
-     * @param index  The index to index into
-     * @param type   The type to index into
-     * @param id     The id of document
-     * @param source The JSON source document
+     * @param index The index to index into
+     * @param type  The type to index into
+     * @param id    The id of document
      */
     public IndexRequest(String index, String type, String id) {
         this.index = index;
@@ -214,7 +213,7 @@ public IndexRequest id(String id) {
      * The source of the JSON document to index.
      */
     byte[] source() {
-        if (sourceUnsafe) {
+        if (sourceUnsafe || sourceOffset > 0) {
             source = Arrays.copyOfRange(source, sourceOffset, sourceLength);
             sourceOffset = 0;
             sourceUnsafe = false;

modules/elasticsearch/src/main/java/org/elasticsearch/action/mlt/MoreLikeThisRequest.java

@@ -372,18 +372,18 @@ public MoreLikeThisRequest searchSource(XContentBuilder builder) {
      * more like this documents.
      */
     public MoreLikeThisRequest searchSource(byte[] searchSource) {
-        return searchSource(searchSource, 0, searchSource.length);
+        return searchSource(searchSource, 0, searchSource.length, false);
     }
 
     /**
      * An optional search source request allowing to control the search request for the
      * more like this documents.
      */
-    public MoreLikeThisRequest searchSource(byte[] searchSource, int offset, int length) {
+    public MoreLikeThisRequest searchSource(byte[] searchSource, int offset, int length, boolean unsafe) {
         this.searchSource = searchSource;
         this.searchSourceOffset = offset;
         this.searchSourceLength = length;
-        this.searchSourceUnsafe = false;
+        this.searchSourceUnsafe = unsafe;
         return this;
     }
 
@@ -403,6 +403,10 @@ public int searchSourceLength() {
         return searchSourceLength;
     }
 
+    public boolean searchSourceUnsafe() {
+        return searchSourceUnsafe;
+    }
+
     /**
      * The search type of the mlt search query.
      */

modules/elasticsearch/src/main/java/org/elasticsearch/action/mlt/TransportMoreLikeThisAction.java

@@ -174,7 +174,7 @@ public class TransportMoreLikeThisAction extends BaseAction<MoreLikeThisRequest,
                         )
                         .listenerThreaded(request.listenerThreaded());
                 if (request.searchSource() != null) {
-                    searchRequest.source(request.searchSource(), request.searchSourceOffset(), request.searchSourceLength());
+                    searchRequest.source(request.searchSource(), request.searchSourceOffset(), request.searchSourceLength(), request.searchSourceUnsafe());
                 }
                 searchAction.execute(searchRequest, new ActionListener<SearchResponse>() {
                     @Override public void onResponse(SearchResponse response) {

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/SearchRequest.java

@@ -267,17 +267,17 @@ public SearchRequest source(XContentBuilder builder) {
      * The search source to execute.
      */
     public SearchRequest source(byte[] source) {
-        return source(source, 0, source.length);
+        return source(source, 0, source.length, false);
     }
 
     /**
      * The search source to execute.
      */
-    public SearchRequest source(byte[] source, int offset, int length) {
+    public SearchRequest source(byte[] source, int offset, int length, boolean unsafe) {
         this.source = source;
         this.sourceOffset = offset;
         this.sourceLength = length;
-        this.sourceUnsafe = false;
+        this.sourceUnsafe = unsafe;
         return this;
     }
 
@@ -346,17 +346,17 @@ public SearchRequest extraSource(String source) {
      * Allows to provide additional source that will be used as well.
      */
     public SearchRequest extraSource(byte[] source) {
-        return extraSource(source, 0, source.length);
+        return extraSource(source, 0, source.length, false);
     }
 
     /**
      * Allows to provide additional source that will be used as well.
      */
-    public SearchRequest extraSource(byte[] source, int offset, int length) {
+    public SearchRequest extraSource(byte[] source, int offset, int length, boolean unsafe) {
         this.extraSource = source;
         this.extraSourceOffset = offset;
         this.extraSourceLength = length;
-        this.extraSourceUnsafe = false;
+        this.extraSourceUnsafe = unsafe;
         return this;
     }
 

modules/elasticsearch/src/main/java/org/elasticsearch/http/netty/NettyHttpRequest.java

@@ -95,6 +95,10 @@ public NettyHttpRequest(org.jboss.netty.handler.codec.http.HttpRequest request)
         return request.getContent().readableBytes();
     }
 
+    @Override public boolean contentUnsafe() {
+        return request.getContent().hasArray();
+    }
+
     @Override public byte[] contentByteArray() {
         if (request.getContent().hasArray()) {
             return request.getContent().array();
@@ -109,7 +113,8 @@ public NettyHttpRequest(org.jboss.netty.handler.codec.http.HttpRequest request)
 
     @Override public int contentByteArrayOffset() {
         if (request.getContent().hasArray()) {
-            return request.getContent().arrayOffset();
+            // get the array offset, and the reader index offset within it
+            return request.getContent().arrayOffset() + request.getContent().readerIndex();
         }
         return 0;
     }

modules/elasticsearch/src/main/java/org/elasticsearch/rest/RestRequest.java

@@ -50,6 +50,11 @@ enum Method {
 
     boolean hasContent();
 
+    /**
+     * Is the byte array content safe or unsafe for usage on other threads
+     */
+    boolean contentUnsafe();
+
     byte[] contentByteArray();
 
     int contentByteArrayOffset();

modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/count/RestCountAction.java

@@ -65,7 +65,7 @@ public class RestCountAction extends BaseRestHandler {
             }
             countRequest.operationThreading(operationThreading);
             if (request.hasContent()) {
-                countRequest.query(request.contentByteArray(), request.contentByteArrayOffset(), request.contentLength());
+                countRequest.query(request.contentByteArray(), request.contentByteArrayOffset(), request.contentLength(), true);
             } else {
                 String source = request.param("source");
                 if (source != null) {

modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/deletebyquery/RestDeleteByQueryAction.java

@@ -55,7 +55,7 @@ public class RestDeleteByQueryAction extends BaseRestHandler {
         deleteByQueryRequest.listenerThreaded(false);
         try {
             if (request.hasContent()) {
-                deleteByQueryRequest.query(request.contentByteArray(), request.contentByteArrayOffset(), request.contentLength());
+                deleteByQueryRequest.query(request.contentByteArray(), request.contentByteArrayOffset(), request.contentLength(), request.contentUnsafe());
             } else {
                 String source = request.param("source");
                 if (source != null) {

modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/index/RestIndexAction.java

@@ -57,7 +57,7 @@ final class CreateHandler implements RestHandler {
 
     @Override public void handleRequest(final RestRequest request, final RestChannel channel) {
         IndexRequest indexRequest = new IndexRequest(request.param("index"), request.param("type"), request.param("id"));
-        indexRequest.source(request.contentByteArray(), request.contentByteArrayOffset(), request.contentLength());
+        indexRequest.source(request.contentByteArray(), request.contentByteArrayOffset(), request.contentLength(), request.contentUnsafe());
         indexRequest.timeout(request.paramAsTime("timeout", IndexRequest.DEFAULT_TIMEOUT));
         String sOpType = request.param("op_type");
         if (sOpType != null) {

modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/mlt/RestMoreLikeThisAction.java

@@ -72,7 +72,7 @@ public class RestMoreLikeThisAction extends BaseRestHandler {
                 mltRequest.searchScroll(new Scroll(parseTimeValue(searchScroll, null)));
             }
             if (request.hasContent()) {
-                mltRequest.searchSource(request.contentByteArray(), request.contentByteArrayOffset(), request.contentLength());
+                mltRequest.searchSource(request.contentByteArray(), request.contentByteArrayOffset(), request.contentLength(), request.contentUnsafe());
             } else {
                 String searchSource = request.param("search_source");
                 if (searchSource != null) {

modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java

@@ -115,7 +115,7 @@ private SearchRequest parseSearchRequest(RestRequest request) {
         SearchRequest searchRequest = new SearchRequest(indices);
         // get the content, and put it in the body
         if (request.hasContent()) {
-            searchRequest.source(request.contentByteArray(), request.contentByteArrayOffset(), request.contentLength());
+            searchRequest.source(request.contentByteArray(), request.contentByteArrayOffset(), request.contentLength(), request.contentUnsafe());
         } else {
             String source = request.param("source");
             if (source != null) {

plugins/transport/memcached/src/main/java/org/elasticsearch/memcached/MemcachedRestRequest.java

@@ -109,6 +109,10 @@ public void setData(byte[] data) {
         return data != null;
     }
 
+    @Override public boolean contentUnsafe() {
+        return false;
+    }
+
     @Override public byte[] contentByteArray() {
         return data;
     }