Updated for Elasticsearch 0.90.6 - Pom file memory settings so it builds and passes. Corrected code for 0.90.6 and fixed issue where >1 thread was trying to write to the same file at the same time. Fixed ATs.

Author: juliakreger

pom.xml

@@ -14,7 +14,7 @@
         <url>https://github.com/crate/elasticsearch-inout-plugin</url>
     </scm>
 	<properties>
-		<elasticsearch.version>0.90.3</elasticsearch.version>
+		<elasticsearch.version>0.90.6</elasticsearch.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
 	<build>
@@ -52,6 +52,7 @@
 					<configuration>
 						<source>1.6</source>
 						<target>1.6</target>
+						<argLine>-X2048m -XX:MaxDirectMemorySize=256m</argLine>
 					</configuration>
 				</plugin>
 			</plugins>

src/main/java/crate/elasticsearch/action/export/AbstractTransportExportAction.java

@@ -149,7 +149,7 @@ protected ShardExportResponse shardOperation(ShardExportRequest request) throws
         SearchShardTarget shardTarget = new SearchShardTarget(clusterService.localNode().id(), request.index(), request.shardId());
         ExportContext context = new ExportContext(0,
             new ShardSearchRequest().types(request.types()).filteringAliases(request.filteringAliases()),
-            shardTarget, indexShard.searcher(), indexService, indexShard, scriptService, cacheRecycler, nodePath);
+            shardTarget, indexShard.acquireSearcher("export"), indexService, indexShard, scriptService, cacheRecycler, nodePath);
         ExportContext.setCurrent(context);
 
         try {

src/main/java/crate/elasticsearch/action/export/ExportRequest.java

@@ -3,7 +3,6 @@
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationRequest;
 import org.elasticsearch.common.Nullable;
-import org.elasticsearch.common.Required;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -59,14 +58,12 @@ protected void beforeStart() {
     public BytesReference source() {
         return source;
     }
-
-    @Required
+ 
     public ExportRequest source(String source) {
         return this.source(new BytesArray(source), false);
     }
 
 
-    @Required
     public ExportRequest source(BytesReference source, boolean unsafe) {
         this.source = source;
         this.querySourceUnsafe = unsafe;

src/main/java/crate/elasticsearch/action/import_/ImportRequest.java

@@ -1,7 +1,6 @@
 package crate.elasticsearch.action.import_;
 
 import org.elasticsearch.action.support.nodes.NodesOperationRequest;
-import org.elasticsearch.common.Required;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -32,12 +31,10 @@ public BytesReference source() {
         return source;
     }
 
-    @Required
     public ImportRequest source(String source) {
         return this.source(new BytesArray(source), false);
     }
 
-    @Required
     public ImportRequest source(BytesReference source, boolean unsafe) {
         this.source = source;
         return this;

src/main/java/crate/elasticsearch/action/searchinto/AbstractTransportSearchIntoAction.java

@@ -168,7 +168,7 @@ protected ShardSearchIntoResponse shardOperation(ShardSearchIntoRequest
                 request.shardId());
         SearchIntoContext context = new SearchIntoContext(0,
             new ShardSearchRequest().types(request.types()).filteringAliases(request.filteringAliases()),
-            shardTarget, indexShard.searcher(), indexService, indexShard, scriptService, cacheRecycler
+            shardTarget, indexShard.acquireSearcher("export"), indexService, indexShard, scriptService, cacheRecycler
         );
         SearchIntoContext.setCurrent(context);
 

src/main/java/crate/elasticsearch/action/searchinto/SearchIntoRequest.java

@@ -3,7 +3,6 @@
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationRequest;
 import org.elasticsearch.common.Nullable;
-import org.elasticsearch.common.Required;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -63,13 +62,11 @@ public BytesReference source() {
         return source;
     }
 
-    @Required
     public SearchIntoRequest source(String source) {
         return this.source(new BytesArray(source), false);
     }
 
 
-    @Required
     public SearchIntoRequest source(BytesReference source, boolean unsafe) {
         this.source = source;
         this.querySourceUnsafe = unsafe;

src/main/java/crate/elasticsearch/export/Exporter.java

@@ -75,7 +75,7 @@ public void check(ExportContext context) throws ExportException {
         }
     }
 
-    public Result execute(ExportContext context) {
+    public synchronized Result execute(ExportContext context) {
         if (context.settings() || context.mappings()) {
             writeSettingsOrMappings(context);
         }

src/main/java/crate/elasticsearch/rest/action/admin/export/RestExportAction.java

@@ -3,7 +3,7 @@
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestStatus.BAD_REQUEST;
 import static org.elasticsearch.rest.RestStatus.OK;
-import static org.elasticsearch.rest.action.support.RestActions.splitTypes;
+import org.elasticsearch.common.Strings;
 
 import java.io.IOException;
 
@@ -52,7 +52,7 @@ protected Action<ExportRequest, ExportResponse, ExportRequestBuilder> action() {
     }
 
     public void handleRequest(final RestRequest request, final RestChannel channel) {
-        ExportRequest exportRequest = new ExportRequest(RestActions.splitIndices(request.param("index")));
+        ExportRequest exportRequest = new ExportRequest(Strings.splitStringByCommaToArray(request.param("index")));
 
         if (request.hasParam("ignore_indices")) {
             exportRequest.ignoreIndices(IgnoreIndices.fromString(request.param("ignore_indices")));
@@ -79,7 +79,7 @@ public void handleRequest(final RestRequest request, final RestChannel channel)
                 }
             }
             exportRequest.routing(request.param("routing"));
-            exportRequest.types(splitTypes(request.param("type")));
+            exportRequest.types(Strings.splitStringByCommaToArray(request.param("type")));
             exportRequest.preference(request.param("preference", "_primary"));
         } catch (Exception e) {
             try {

src/main/java/crate/elasticsearch/rest/action/admin/searchinto/RestSearchIntoAction.java

@@ -3,8 +3,8 @@
 import static org.elasticsearch.rest.RestRequest.Method.POST;
 import static org.elasticsearch.rest.RestStatus.BAD_REQUEST;
 import static org.elasticsearch.rest.RestStatus.OK;
-import static org.elasticsearch.rest.action.support.RestActions.splitTypes;
-
+/** import static org.elasticsearch.rest.action.support.RestActions; **/
+import org.elasticsearch.common.Strings;
 import java.io.IOException;
 
 import org.elasticsearch.action.Action;
@@ -55,7 +55,7 @@ protected Action<SearchIntoRequest, SearchIntoResponse, SearchIntoRequestBuilder
     public void handleRequest(final RestRequest request,
             final RestChannel channel) {
         SearchIntoRequest searchIntoRequest = new SearchIntoRequest(
-                RestActions.splitIndices(request.param("index")));
+                Strings.splitStringByCommaToArray(request.param("index")));
 
         if (request.hasParam("ignore_indices")) {
             searchIntoRequest.ignoreIndices(IgnoreIndices.fromString(
@@ -90,7 +90,7 @@ public void handleRequest(final RestRequest request,
                 }
             }
             searchIntoRequest.routing(request.param("routing"));
-            searchIntoRequest.types(splitTypes(request.param("type")));
+            searchIntoRequest.types(Strings.splitStringByCommaToArray(request.param("type")));
             searchIntoRequest.preference(request.param("preference",
                     "_primary"));
         } catch (Exception e) {

src/test/python/reindex.rst

@@ -54,7 +54,7 @@ be closed first and then reopened::
     >>> post("/test/_close", {})
     {"ok":true,"acknowledged":true}
     >>> put("/test/_settings", {"analysis": {"analyzer": {"myan": {"type": "stop", "stopwords": ["nice"]}}}})
-    {"ok":true}
+    {"ok":true,"acknowledged":true}
     >>> post("/test/_open", {})
     {"ok":true,"acknowledged":true}
     >>> refresh()