Update to ElasticSearch 0.90.7

pom.xml

@@ -5,7 +5,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>crate</groupId>
 	<artifactId>elasticsearch-inout-plugin</artifactId>
-	<version>0.5.0</version>
+	<version>0.90.7</version>
     <packaging>jar</packaging>
     <description>An Elasticsearch plugin which provides the ability to export and import data by query on server side.</description>
     <scm>
@@ -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.7</elasticsearch.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
 	<build>

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("inout-plugin"), 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;
@@ -60,13 +59,11 @@ 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("inout-plugin"), 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/rest/action/admin/export/RestExportAction.java

@@ -3,7 +3,6 @@
 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 java.io.IOException;
 
@@ -12,6 +11,7 @@
 import org.elasticsearch.action.support.IgnoreIndices;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationThreading;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
@@ -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.commaDelimitedListToStringArray(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.commaDelimitedListToStringArray(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,7 +3,6 @@
 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 java.io.IOException;
 
@@ -12,6 +11,7 @@
 import org.elasticsearch.action.support.IgnoreIndices;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationThreading;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
@@ -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.commaDelimitedListToStringArray(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.commaDelimitedListToStringArray(request.param("type")));
             searchIntoRequest.preference(request.param("preference",
                     "_primary"));
         } catch (Exception e) {

src/test/java/crate/elasticsearch/module/export/test/RestExportActionTest.java

@@ -1,32 +1,24 @@
 package crate.elasticsearch.module.export.test;
 
-import static com.github.tlrx.elasticsearch.test.EsSetup.createIndex;
-import static com.github.tlrx.elasticsearch.test.EsSetup.deleteAll;
-import static com.github.tlrx.elasticsearch.test.EsSetup.fromClassPath;
-import static com.github.tlrx.elasticsearch.test.EsSetup.index;
-
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.FileReader;
-import java.io.IOException;
-import java.util.Date;
-import java.util.List;
-import java.util.Map;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
+import com.github.tlrx.elasticsearch.test.EsSetup;
+import crate.elasticsearch.action.export.ExportAction;
+import crate.elasticsearch.action.export.ExportRequest;
+import crate.elasticsearch.action.export.ExportResponse;
+import crate.elasticsearch.module.AbstractRestActionTest;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.indices.IndexMissingException;
 import org.junit.Test;
 
-import com.github.tlrx.elasticsearch.test.EsSetup;
+import java.io.*;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
-import crate.elasticsearch.action.export.ExportAction;
-import crate.elasticsearch.action.export.ExportRequest;
-import crate.elasticsearch.action.export.ExportResponse;
-import crate.elasticsearch.module.AbstractRestActionTest;
+import static com.github.tlrx.elasticsearch.test.EsSetup.*;
 
 public class RestExportActionTest extends AbstractRestActionTest {
 
@@ -231,17 +223,40 @@ public void testOutputFileAndOutputCommand() {
      */
     @Test
     public void testForceOverwrite() {
-        String filename = "/tmp/filename.export";
-        ExportResponse response = executeExportRequest("{\"output_file\": \"" + filename +
-                "\", \"fields\": [\"name\"], \"force_overwrite\": \"true\"}");
+        String filename = "/tmp/filename-" + System.currentTimeMillis() + "-${index}-${shard}.export";
+        int lineCount= 0;
+        {
+            ExportResponse response = executeExportRequest("{\"output_file\": \"" + filename +
+                    "\", \"fields\": [\"name\"], \"force_overwrite\": \"false\"}");
+
+            List<Map<String, Object>> infos = getExports(response);
+            assertEquals(2, infos.size());
+            String out1 = infos.get(0).get("output_file").toString();
+            String out2 = infos.get(1).get("output_file").toString();
+            List<String> lines = readLines(out1);
+            lines.addAll(readLines(out2));
+            assertTrue(lines.size() > 0);
+            lineCount = lines.size();
+        }
+        {
+            ExportResponse response = executeExportRequest("{\"output_file\": \"" + filename +
+                    "\", \"fields\": [\"name\"], \"force_overwrite\": \"false\"}");
 
-        List<Map<String, Object>> infos = getExports(response);
-        assertEquals(2, infos.size());
-        assertEquals("/tmp/filename.export", infos.get(0).get("output_file").toString());
-        assertEquals("/tmp/filename.export", infos.get(1).get("output_file").toString());
-        List<String> lines = readLines(filename);
-        assertEquals(2, lines.size());
-        assertEquals("{\"name\":\"bike\"}", lines.get(0));
+            List<Map<String, Object>> infos = getExports(response);
+            assertEquals(0, infos.size());
+        }
+        {
+            ExportResponse response = executeExportRequest("{\"output_file\": \"" + filename +
+                    "\", \"fields\": [\"name\"], \"force_overwrite\": \"true\"}");
+
+            List<Map<String, Object>> infos = getExports(response);
+            assertEquals(2, infos.size());
+            String out1 = infos.get(0).get("output_file").toString();
+            String out2 = infos.get(1).get("output_file").toString();
+            List<String> lines = readLines(out1);
+            lines.addAll(readLines(out2));
+            assertEquals(lineCount, lines.size());
+        }
     }
 
     /**
@@ -286,7 +301,7 @@ public void testWithMultipleNodes() {
         esSetup2 = new EsSetup();
         esSetup2.execute(index("users", "d").withSource("{\"name\": \"motorbike\"}"));
         esSetup2.client().admin().cluster().prepareHealth().setWaitForGreenStatus().
-            setWaitForNodes("2").setWaitForRelocatingShards(0).execute().actionGet();
+                setWaitForNodes("2").setWaitForRelocatingShards(0).execute().actionGet();
 
         // Do export request
         String source = "{\"output_cmd\": \"cat\", \"fields\": [\"name\"]}";
@@ -308,7 +323,7 @@ public void testWithMultipleNodes() {
     public void testWithQuery() {
         ExportResponse response = executeExportRequest(
                 "{\"output_file\": \"/tmp/query-${shard}.json\", \"fields\": [\"name\"], " +
-                "\"query\": {\"match\": {\"name\":\"bus\"}}, \"force_overwrite\": true}");
+                        "\"query\": {\"match\": {\"name\":\"bus\"}}, \"force_overwrite\": true}");
 
         assertEquals(0, response.getFailedShards());
         List<Map<String, Object>> infos = getExports(response);
@@ -404,10 +419,10 @@ public void testTimestampNotStored() {
      * as a field.
      */
     @Test
-    public void testTimestampStored(){
+    public void testTimestampStored() {
         esSetup.execute(deleteAll(), createIndex("tsstored").withSettings(
                 fromClassPath("essetup/settings/test_a.json")).withMapping("d",
-                        "{\"d\": {\"_timestamp\": {\"enabled\": true, \"store\": \"yes\"}}}"));
+                "{\"d\": {\"_timestamp\": {\"enabled\": true, \"store\": \"yes\"}}}"));
         Client client = esSetup.client();
         client.prepareIndex("tsstored", "d", "1").setSource(
                 "field1", "value1").setTimestamp("123").execute().actionGet();
@@ -438,7 +453,7 @@ public void testTTLNotEnabled() {
     public void testTTLEnabled() {
         esSetup.execute(deleteAll(), createIndex("ttlenabled").withSettings(
                 fromClassPath("essetup/settings/test_a.json")).withMapping("d",
-                        "{\"d\": {\"_ttl\": {\"enabled\": true, \"default\": \"1d\"}}}"));
+                "{\"d\": {\"_ttl\": {\"enabled\": true, \"default\": \"1d\"}}}"));
         Client client = esSetup.client();
         client.prepareIndex("ttlenabled", "d", "1").setSource("field1", "value1").execute().actionGet();
         client.admin().indices().prepareRefresh().execute().actionGet();
@@ -449,7 +464,7 @@ public void testTTLEnabled() {
         List<Map<String, Object>> infos = getExports(response);
         String stdout = infos.get(1).get("stdout").toString();
         assertTrue(stdout.startsWith("{\"_id\":\"1\",\"_ttl\":"));
-        String lsplit  = stdout.substring(18);
+        String lsplit = stdout.substring(18);
         long ttl = Long.valueOf(lsplit.substring(0, lsplit.length() - 2));
         long diff = ttl - now.getTime();
         assertTrue(diff < 86400000 && diff > 86390000);
@@ -496,7 +511,7 @@ public void testExportRelativeFilename() {
         esSetup2 = new EsSetup();
         esSetup2.execute(index("users", "d").withSource("{\"name\": \"motorbike\"}"));
         esSetup2.client().admin().cluster().prepareHealth().setWaitForGreenStatus().
-            setWaitForNodes("2").setWaitForRelocatingShards(0).execute().actionGet();
+                setWaitForNodes("2").setWaitForRelocatingShards(0).execute().actionGet();
 
         ExportResponse response = executeExportRequest(
                 "{\"output_file\": \"export.${shard}.${index}.json\", \"fields\": [\"name\", \"_id\"], \"force_overwrite\": true}");
@@ -673,7 +688,7 @@ private List<String> readLines(String filename) {
     }
 
     private void assertShardInfoCommand(Map<String, Object> map, String index,
-            int exitcode, String stdout, String stderr, String cmd) {
+                                        int exitcode, String stdout, String stderr, String cmd) {
         assertEquals(index, map.get("index"));
         assertEquals(exitcode, map.get("exitcode"));
         assertEquals(stderr, map.get("stderr"));

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()