elasticsearch 0.90.3 compatibility

mfussenegger · mfussenegger · commit 7bccc76babe6 · 2013-08-19T15:21:52.000+02:00
note that this change is incompatible with 0.90.2 or other
earlier versions of elasticsearch
diff --git a/CHANGES.txt b/CHANGES.txt
@@ -5,6 +5,10 @@ Changes for Elasticsearch InOut Plugin
 Unreleased
 ==========
 
+ - elasticsearch 0.90.3 compatibility.
+   NOTE: This is incompatible with 0.90.2 or other earlier versions of
+   elasticsearch
+
 2013/07/08 0.5.0
 ================
 
diff --git a/pom.xml b/pom.xml
@@ -14,7 +14,7 @@
         <url>https://github.com/crate/elasticsearch-inout-plugin</url>
     </scm>
 	<properties>
-		<elasticsearch.version>0.90.2</elasticsearch.version>
+		<elasticsearch.version>0.90.3</elasticsearch.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
 	<build>
diff --git a/src/main/java/crate/elasticsearch/action/dump/TransportDumpAction.java b/src/main/java/crate/elasticsearch/action/dump/TransportDumpAction.java
@@ -3,6 +3,7 @@
 import crate.elasticsearch.action.dump.parser.DumpParser;
 import crate.elasticsearch.action.export.AbstractTransportExportAction;
 import crate.elasticsearch.export.Exporter;
+import org.elasticsearch.cache.recycler.CacheRecycler;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
@@ -21,9 +22,10 @@ public class TransportDumpAction extends AbstractTransportExportAction {
     @Inject
     public TransportDumpAction(Settings settings, ThreadPool threadPool, ClusterService clusterService,
                                TransportService transportService, IndicesService indicesService,
-                               ScriptService scriptService, DumpParser dumpParser, Exporter exporter,
-                               NodeEnvironment nodeEnv) {
-        super(settings, threadPool, clusterService, transportService, indicesService, scriptService, dumpParser, exporter, nodeEnv);
+                               ScriptService scriptService, CacheRecycler cacheRecycler,
+                               DumpParser dumpParser, Exporter exporter, NodeEnvironment nodeEnv) {
+        super(settings, threadPool, clusterService, transportService, indicesService, scriptService,
+            cacheRecycler, dumpParser, exporter, nodeEnv);
     }
 
     @Override
diff --git a/src/main/java/crate/elasticsearch/action/export/AbstractTransportExportAction.java b/src/main/java/crate/elasticsearch/action/export/AbstractTransportExportAction.java
@@ -7,6 +7,7 @@
 import org.elasticsearch.action.support.DefaultShardOperationFailedException;
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationFailedException;
 import org.elasticsearch.action.support.broadcast.TransportBroadcastOperationAction;
+import org.elasticsearch.cache.recycler.CacheRecycler;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -50,15 +51,19 @@ public abstract class AbstractTransportExportAction extends TransportBroadcastOp
 
     private final Exporter exporter;
 
+    private final CacheRecycler cacheRecycler;
+
     private String nodePath;
 
     public AbstractTransportExportAction(Settings settings, ThreadPool threadPool, ClusterService clusterService,
                                          TransportService transportService, IndicesService indicesService,
-                                         ScriptService scriptService, IExportParser exportParser, Exporter exporter,
+                                         ScriptService scriptService, CacheRecycler cacheRecycler,
+                                         IExportParser exportParser, Exporter exporter,
                                          NodeEnvironment nodeEnv) {
         super(settings, threadPool, clusterService, transportService);
         this.indicesService = indicesService;
         this.scriptService = scriptService;
+        this.cacheRecycler = cacheRecycler;
         this.exportParser = exportParser;
         this.exporter = exporter;
         File[] paths = nodeEnv.nodeDataLocations();
@@ -142,7 +147,9 @@ protected ShardExportResponse shardOperation(ShardExportRequest request) throws
         IndexShard indexShard = indexService.shardSafe(request.shardId());
 
         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, nodePath);
+        ExportContext context = new ExportContext(0,
+            new ShardSearchRequest().types(request.types()).filteringAliases(request.filteringAliases()),
+            shardTarget, indexShard.searcher(), indexService, indexShard, scriptService, cacheRecycler, nodePath);
         ExportContext.setCurrent(context);
 
         try {
diff --git a/src/main/java/crate/elasticsearch/action/export/ExportContext.java b/src/main/java/crate/elasticsearch/action/export/ExportContext.java
@@ -3,6 +3,7 @@
 import crate.elasticsearch.export.Output;
 import crate.elasticsearch.export.OutputCommand;
 import crate.elasticsearch.export.OutputFile;
+import org.elasticsearch.cache.recycler.CacheRecycler;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.service.IndexService;
@@ -34,8 +35,10 @@ public class ExportContext extends SearchContext {
     private boolean mappings = false;
     private boolean settings = false;
 
-    public ExportContext(long id, ShardSearchRequest request, SearchShardTarget shardTarget, Engine.Searcher engineSearcher, IndexService indexService, IndexShard indexShard, ScriptService scriptService, String nodePath) {
-        super(id, request, shardTarget, engineSearcher, indexService, indexShard, scriptService);
+    public ExportContext(long id, ShardSearchRequest request, SearchShardTarget shardTarget,
+                         Engine.Searcher engineSearcher, IndexService indexService, IndexShard indexShard,
+                         ScriptService scriptService, CacheRecycler cacheRecycler, String nodePath) {
+        super(id, request, shardTarget, engineSearcher, indexService, indexShard, scriptService, cacheRecycler);
         this.nodePath = nodePath;
     }
 
diff --git a/src/main/java/crate/elasticsearch/action/export/TransportExportAction.java b/src/main/java/crate/elasticsearch/action/export/TransportExportAction.java
@@ -2,6 +2,7 @@
 
 import crate.elasticsearch.action.export.parser.ExportParser;
 import crate.elasticsearch.export.Exporter;
+import org.elasticsearch.cache.recycler.CacheRecycler;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
@@ -20,9 +21,10 @@ public class TransportExportAction extends AbstractTransportExportAction {
     @Inject
     public TransportExportAction(Settings settings, ThreadPool threadPool, ClusterService clusterService,
                                  TransportService transportService, IndicesService indicesService,
-                                 ScriptService scriptService, ExportParser exportParser, Exporter exporter,
-                                 NodeEnvironment nodeEnv) {
-        super(settings, threadPool, clusterService, transportService, indicesService, scriptService, exportParser, exporter, nodeEnv);
+                                 ScriptService scriptService, CacheRecycler cacheRecycler,
+                                 ExportParser exportParser, Exporter exporter, NodeEnvironment nodeEnv) {
+        super(settings, threadPool, clusterService, transportService, indicesService, scriptService,
+            cacheRecycler, exportParser, exporter, nodeEnv);
     }
 
     @Override
diff --git a/src/main/java/crate/elasticsearch/action/reindex/TransportReindexAction.java b/src/main/java/crate/elasticsearch/action/reindex/TransportReindexAction.java
@@ -1,5 +1,6 @@
 package crate.elasticsearch.action.reindex;
 
+import org.elasticsearch.cache.recycler.CacheRecycler;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
@@ -15,10 +16,10 @@ public class TransportReindexAction extends AbstractTransportSearchIntoAction {
 
     @Inject
     public TransportReindexAction(Settings settings, ThreadPool threadPool,
-            ClusterService clusterService, TransportService transportService,
+            ClusterService clusterService, TransportService transportService, CacheRecycler cacheRecycler,
             IndicesService indicesService, ScriptService scriptService,
             ReindexParser parser, Writer writer) {
-        super(settings, threadPool, clusterService, transportService, indicesService,
+        super(settings, threadPool, clusterService, transportService, cacheRecycler, indicesService,
                 scriptService, parser, writer);
     }
 
diff --git a/src/main/java/crate/elasticsearch/action/searchinto/AbstractTransportSearchIntoAction.java b/src/main/java/crate/elasticsearch/action/searchinto/AbstractTransportSearchIntoAction.java
@@ -13,6 +13,7 @@
 import org.elasticsearch.action.support.DefaultShardOperationFailedException;
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationFailedException;
 import org.elasticsearch.action.support.broadcast.TransportBroadcastOperationAction;
+import org.elasticsearch.cache.recycler.CacheRecycler;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -53,16 +54,19 @@ public abstract class AbstractTransportSearchIntoAction extends
 
     private final ISearchIntoParser parser;
 
+    private final CacheRecycler cacheRecycler;
+
     private final Writer writer;
 
     @Inject
     public AbstractTransportSearchIntoAction(Settings settings,
             ThreadPool threadPool, ClusterService clusterService,
-            TransportService transportService,
+            TransportService transportService, CacheRecycler cacheRecycler,
             IndicesService indicesService, ScriptService scriptService,
             ISearchIntoParser parser, Writer writer) {
         super(settings, threadPool, clusterService, transportService);
         this.indicesService = indicesService;
+        this.cacheRecycler = cacheRecycler;
         this.scriptService = scriptService;
         this.parser = parser;
         this.writer = writer;
@@ -163,11 +167,9 @@ protected ShardSearchIntoResponse shardOperation(ShardSearchIntoRequest
                 clusterService.localNode().id(), request.index(),
                 request.shardId());
         SearchIntoContext context = new SearchIntoContext(0,
-                new ShardSearchRequest().types(
-                        request.types()).filteringAliases(
-                        request.filteringAliases()), shardTarget,
-                indexShard.searcher(), indexService, indexShard,
-                scriptService);
+            new ShardSearchRequest().types(request.types()).filteringAliases(request.filteringAliases()),
+            shardTarget, indexShard.searcher(), indexService, indexShard, scriptService, cacheRecycler
+        );
         SearchIntoContext.setCurrent(context);
 
         try {
diff --git a/src/main/java/crate/elasticsearch/action/searchinto/SearchIntoContext.java b/src/main/java/crate/elasticsearch/action/searchinto/SearchIntoContext.java
@@ -1,5 +1,6 @@
 package crate.elasticsearch.action.searchinto;
 
+import org.elasticsearch.cache.recycler.CacheRecycler;
 import org.elasticsearch.common.collect.ImmutableList;
 import org.elasticsearch.common.collect.Lists;
 import org.elasticsearch.common.transport.InetSocketTransportAddress;
@@ -36,9 +37,9 @@ public Map<String, String> outputNames() {
     public SearchIntoContext(long id, ShardSearchRequest request,
             SearchShardTarget shardTarget, Engine.Searcher engineSearcher,
             IndexService indexService, IndexShard indexShard,
-            ScriptService scriptService) {
+            ScriptService scriptService, CacheRecycler cacheRecycler) {
         super(id, request, shardTarget, engineSearcher, indexService,
-                indexShard, scriptService);
+                indexShard, scriptService, cacheRecycler);
     }
 
     public String targetType() {
diff --git a/src/main/java/crate/elasticsearch/action/searchinto/TransportSearchIntoAction.java b/src/main/java/crate/elasticsearch/action/searchinto/TransportSearchIntoAction.java
@@ -1,5 +1,6 @@
 package crate.elasticsearch.action.searchinto;
 
+import org.elasticsearch.cache.recycler.CacheRecycler;
 import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
@@ -20,10 +21,11 @@ public class TransportSearchIntoAction extends AbstractTransportSearchIntoAction
     @Inject
     public TransportSearchIntoAction(Settings settings,
             ThreadPool threadPool, ClusterService clusterService,
-            TransportService transportService,
+            TransportService transportService, CacheRecycler cacheRecycler,
             IndicesService indicesService, ScriptService scriptService,
             SearchIntoParser parser, Writer writer) {
-        super(settings, threadPool, clusterService, transportService, indicesService, scriptService, parser, writer);
+        super(settings, threadPool, clusterService, transportService, cacheRecycler, indicesService,
+            scriptService, parser, writer);
     }
 
     @Override
diff --git a/src/main/java/crate/elasticsearch/export/ExportCollector.java b/src/main/java/crate/elasticsearch/export/ExportCollector.java
@@ -15,7 +15,7 @@
 import org.apache.lucene.search.Collector;
 import org.apache.lucene.search.Scorer;
 import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.io.stream.CachedStreamOutput;
+import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.text.StringAndBytesText;
 import org.elasticsearch.common.text.Text;
 import org.elasticsearch.common.xcontent.ToXContent;
@@ -171,14 +171,12 @@ public void collect(int doc) throws IOException {
 
         searchHit.shardTarget(context.shardTarget());
         exportFields.hit(searchHit);
-        CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
-        XContentBuilder builder = new XContentBuilder(XContentFactory
-                .xContent(XContentType.JSON), cachedEntry.bytes());
+        BytesStreamOutput os = new BytesStreamOutput();
+        XContentBuilder builder = new XContentBuilder(XContentFactory.xContent(XContentType.JSON), os);
         exportFields.toXContent(builder, ToXContent.EMPTY_PARAMS);
         builder.flush();
-        BytesReference bytes = cachedEntry.bytes().bytes();
+        BytesReference bytes = os.bytes();
         out.write(bytes.array(), bytes.arrayOffset(), bytes.length());
-        CachedStreamOutput.pushEntry(cachedEntry);
         out.write('\n');
         out.flush();
         numExported++;
diff --git a/src/main/java/crate/elasticsearch/import_/Importer.java b/src/main/java/crate/elasticsearch/import_/Importer.java
@@ -370,7 +370,7 @@ private ImmutableMap<String, IndexMetaData> getIndexMetaData(Set<String> indexes
                 .filteredIndices(indexes.toArray(new String[indexes.size()]));
         clusterStateRequest.listenerThreaded(false);
         ClusterStateResponse response = client.admin().cluster().state(clusterStateRequest).actionGet();
-        return response.getState().metaData().indices();
+        return ImmutableMap.copyOf(response.getState().metaData().indices());
     }
 
 
diff --git a/src/test/java/crate/elasticsearch/module/import_/test/RestImportActionTest.java b/src/test/java/crate/elasticsearch/module/import_/test/RestImportActionTest.java
@@ -326,7 +326,8 @@ public void testMappings() {
         executeImportRequest("{\"directory\": \"" + path + "\", \"mappings\": true}");
 
         ClusterStateRequest clusterStateRequest = Requests.clusterStateRequest().filteredIndices("index1");
-        ImmutableMap<String, MappingMetaData> mappings = esSetup.client().admin().cluster().state(clusterStateRequest).actionGet().getState().metaData().index("index1").getMappings();
+        ImmutableMap<String, MappingMetaData> mappings = ImmutableMap.copyOf(
+            esSetup.client().admin().cluster().state(clusterStateRequest).actionGet().getState().metaData().index("index1").getMappings());
         assertEquals("{\"1\":{\"_timestamp\":{\"enabled\":true,\"store\":true},\"_ttl\":{\"enabled\":true,\"default\":86400000},\"properties\":{\"name\":{\"type\":\"string\",\"store\":true}}}}",
                 mappings.get("1").source().toString());
     }
