diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/broadcast/TransportBroadcastPingAction.java b/modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/broadcast/TransportBroadcastPingAction.java index 7304a5513c3..e1572739b08 100644 --- a/modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/broadcast/TransportBroadcastPingAction.java +++ b/modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/broadcast/TransportBroadcastPingAction.java @@ -65,7 +65,7 @@ public class TransportBroadcastPingAction extends TransportBroadcastOperationAct } @Override protected GroupShardsIterator shards(BroadcastPingRequest request, ClusterState clusterState) { - return clusterService.operationRouting().searchShards(clusterState, request.indices(), request.queryHint(), null); + return clusterService.operationRouting().searchShards(clusterState, request.indices(), request.queryHint(), null, null); } @Override protected BroadcastPingResponse newResponse(BroadcastPingRequest request, AtomicReferenceArray shardsResponses, ClusterState clusterState) { diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/action/count/TransportCountAction.java b/modules/elasticsearch/src/main/java/org/elasticsearch/action/count/TransportCountAction.java index eed72e8a3d8..a299bf1fa2f 100644 --- a/modules/elasticsearch/src/main/java/org/elasticsearch/action/count/TransportCountAction.java +++ b/modules/elasticsearch/src/main/java/org/elasticsearch/action/count/TransportCountAction.java @@ -83,7 +83,7 @@ public class TransportCountAction extends TransportBroadcastOperationAction_local to prefer local shards, _primary to execute only on primary shards, or + * a custom value, which guarantees that the same order will be used across different requests. + */ + public GetRequest preference(String preference) { + this.preference = preference; + return this; + } /** * Explicitly specify the fields that will be returned. By default, the _source diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/action/search/SearchRequest.java b/modules/elasticsearch/src/main/java/org/elasticsearch/action/search/SearchRequest.java index fff986e951c..dcb5ab7be07 100644 --- a/modules/elasticsearch/src/main/java/org/elasticsearch/action/search/SearchRequest.java +++ b/modules/elasticsearch/src/main/java/org/elasticsearch/action/search/SearchRequest.java @@ -71,6 +71,7 @@ public class SearchRequest implements ActionRequest { @Nullable private String queryHint; @Nullable private String routing; + @Nullable private String preference; private byte[] source; private int sourceOffset; @@ -220,6 +221,20 @@ public class SearchRequest implements ActionRequest { return this; } + /** + * Sets the preference to execute the search. Defaults to randomize across shards. Can be set to + * _local to prefer local shards, _primary to execute only on primary shards, or + * a custom value, which guarantees that the same order will be used across different requests. + */ + public SearchRequest preference(String preference) { + this.preference = preference; + return this; + } + + public String preference() { + return this.preference; + } + /** * The search type to execute, defaults to {@link SearchType#DEFAULT}. */ @@ -509,6 +524,9 @@ public class SearchRequest implements ActionRequest { if (in.readBoolean()) { routing = in.readUTF(); } + if (in.readBoolean()) { + preference = in.readUTF(); + } if (in.readBoolean()) { scroll = readScroll(in); @@ -567,6 +585,12 @@ public class SearchRequest implements ActionRequest { out.writeBoolean(true); out.writeUTF(routing); } + if (preference == null) { + out.writeBoolean(false); + } else { + out.writeBoolean(true); + out.writeUTF(preference); + } if (scroll == null) { out.writeBoolean(false); diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java b/modules/elasticsearch/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java index 547da7b0e27..bdfc84486bd 100644 --- a/modules/elasticsearch/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java +++ b/modules/elasticsearch/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java @@ -85,7 +85,7 @@ public class TransportSearchAction extends BaseAction_local to prefer local shards, _primary to execute only on primary shards, or + * a custom value, which guarantees that the same order will be used across different requests. + */ + public GetRequestBuilder setPreference(String preference) { + request.preference(preference); + return this; + } + /** * Explicitly specify the fields that will be returned. By default, the _source * field will be returned. diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/client/action/search/SearchRequestBuilder.java b/modules/elasticsearch/src/main/java/org/elasticsearch/client/action/search/SearchRequestBuilder.java index 8b7721903df..ae139ca1a5a 100644 --- a/modules/elasticsearch/src/main/java/org/elasticsearch/client/action/search/SearchRequestBuilder.java +++ b/modules/elasticsearch/src/main/java/org/elasticsearch/client/action/search/SearchRequestBuilder.java @@ -152,6 +152,16 @@ public class SearchRequestBuilder extends BaseRequestBuilder_local to prefer local shards, _primary to execute only on primary shards, or + * a custom value, which guarantees that the same order will be used across different requests. + */ + public SearchRequestBuilder setPreference(String preference) { + request.preference(preference); + return this; + } + /** * Controls the the search operation threading model. */ diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/IndexShardRoutingTable.java b/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/IndexShardRoutingTable.java index 92998f0868d..76e54a725ea 100644 --- a/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/IndexShardRoutingTable.java +++ b/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/IndexShardRoutingTable.java @@ -28,6 +28,7 @@ import org.elasticsearch.common.util.concurrent.jsr166y.ThreadLocalRandom; import org.elasticsearch.index.shard.ShardId; import java.io.IOException; +import java.util.ArrayList; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; @@ -99,6 +100,9 @@ public class IndexShardRoutingTable implements Iterable { return count; } + /** + * Returns a regular shard iterator. + */ public ShardIterator shardsIt() { return new PlainShardIterator(shardId, shards); } @@ -107,6 +111,43 @@ public class IndexShardRoutingTable implements Iterable { return new PlainShardIterator(shardId, shards, index); } + /** + * Returns an iterator only on the primary shard. + */ + public ShardIterator primaryShardIt() { + ShardRouting primary = primaryShard(); + if (primary == null) { + return new PlainShardIterator(shardId, ImmutableList.of()); + } + return new PlainShardIterator(shardId, ImmutableList.of(primary)); + } + + /** + * Prefers execution on the local node if applicable. + */ + public ShardIterator preferLocalShardsIt(String nodeId) { + ArrayList ordered = new ArrayList(this.shards.size()); + // fill it in a randomized fashion + int index = counter.getAndIncrement(); + for (int i = 0; i < this.shards.size(); i++) { + int loc = (index + i) % this.shards.size(); + ordered.add(this.shards.get(loc)); + } + // find the local one, and push it upfront + for (int i = 0; i < ordered.size(); i++) { + ShardRouting current = ordered.get(i); + if (nodeId.equals(current.currentNodeId())) { + ordered.set(i, ordered.get(0)); + ordered.set(0, current); + break; + } + } + return new PlainShardIterator(shardId, ordered); + } + + /** + * Returns a random shards iterator. + */ public ShardIterator shardsRandomIt() { return new PlainShardIterator(shardId, shards, counter.getAndIncrement()); } diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/OperationRouting.java b/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/OperationRouting.java index 9e83e685c12..69b11302589 100644 --- a/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/OperationRouting.java +++ b/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/OperationRouting.java @@ -37,9 +37,9 @@ public interface OperationRouting { GroupShardsIterator broadcastDeleteShards(ClusterState clusterState, String index) throws IndexMissingException, IndexShardMissingException; - ShardIterator getShards(ClusterState clusterState, String index, String type, String id, @Nullable String routing) throws IndexMissingException, IndexShardMissingException; + ShardIterator getShards(ClusterState clusterState, String index, String type, String id, @Nullable String routing, @Nullable String preference) throws IndexMissingException, IndexShardMissingException; GroupShardsIterator deleteByQueryShards(ClusterState clusterState, String index, @Nullable String routing) throws IndexMissingException; - GroupShardsIterator searchShards(ClusterState clusterState, String[] indices, @Nullable String queryHint, @Nullable String routing) throws IndexMissingException; + GroupShardsIterator searchShards(ClusterState clusterState, String[] indices, @Nullable String queryHint, @Nullable String routing, @Nullable String preference) throws IndexMissingException; } diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/hash/djb/DjbHashFunction.java b/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/hash/djb/DjbHashFunction.java index 8886b4832c6..f5a5594383b 100644 --- a/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/hash/djb/DjbHashFunction.java +++ b/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/hash/djb/DjbHashFunction.java @@ -26,16 +26,20 @@ import org.elasticsearch.cluster.routing.operation.hash.HashFunction; */ public class DjbHashFunction implements HashFunction { - @Override public int hash(String routing) { + public static int DJB_HASH(String value) { long hash = 5381; - for (int i = 0; i < routing.length(); i++) { - hash = ((hash << 5) + hash) + routing.charAt(i); + for (int i = 0; i < value.length(); i++) { + hash = ((hash << 5) + hash) + value.charAt(i); } return (int) hash; } + @Override public int hash(String routing) { + return DJB_HASH(routing); + } + @Override public int hash(String type, String id) { long hash = 5381; diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/plain/PlainOperationRouting.java b/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/plain/PlainOperationRouting.java index b1d44d4a9c3..08478815856 100644 --- a/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/plain/PlainOperationRouting.java +++ b/modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/operation/plain/PlainOperationRouting.java @@ -27,6 +27,7 @@ import org.elasticsearch.cluster.routing.IndexShardRoutingTable; import org.elasticsearch.cluster.routing.ShardIterator; import org.elasticsearch.cluster.routing.operation.OperationRouting; import org.elasticsearch.cluster.routing.operation.hash.HashFunction; +import org.elasticsearch.cluster.routing.operation.hash.djb.DjbHashFunction; import org.elasticsearch.common.Nullable; import org.elasticsearch.common.component.AbstractComponent; import org.elasticsearch.common.inject.Inject; @@ -65,8 +66,8 @@ public class PlainOperationRouting extends AbstractComponent implements Operatio return shards(clusterState, index, type, id, routing).shardsIt(); } - @Override public ShardIterator getShards(ClusterState clusterState, String index, String type, String id, @Nullable String routing) throws IndexMissingException, IndexShardMissingException { - return shards(clusterState, index, type, id, routing).shardsRandomIt(); + @Override public ShardIterator getShards(ClusterState clusterState, String index, String type, String id, @Nullable String routing, @Nullable String preference) throws IndexMissingException, IndexShardMissingException { + return preferenceShardIterator(shards(clusterState, index, type, id, routing), clusterState.nodes().localNodeId(), preference); } @Override public GroupShardsIterator broadcastDeleteShards(ClusterState clusterState, String index) throws IndexMissingException { @@ -97,7 +98,7 @@ public class PlainOperationRouting extends AbstractComponent implements Operatio return new GroupShardsIterator(set); } - @Override public GroupShardsIterator searchShards(ClusterState clusterState, String[] indices, @Nullable String queryHint, @Nullable String routing) throws IndexMissingException { + @Override public GroupShardsIterator searchShards(ClusterState clusterState, String[] indices, @Nullable String queryHint, @Nullable String routing, @Nullable String preference) throws IndexMissingException { if (indices == null || indices.length == 0) { indices = clusterState.metaData().concreteAllIndices(); } @@ -119,7 +120,7 @@ public class PlainOperationRouting extends AbstractComponent implements Operatio throw new IndexShardMissingException(new ShardId(index, shardId)); } // we might get duplicates, but that's ok, they will override one another - set.add(indexShard.shardsRandomIt()); + set.add(preferenceShardIterator(indexShard, clusterState.nodes().localNodeId(), preference)); } } return new GroupShardsIterator(set); @@ -129,13 +130,27 @@ public class PlainOperationRouting extends AbstractComponent implements Operatio for (String index : indices) { IndexRoutingTable indexRouting = indexRoutingTable(clusterState, index); for (IndexShardRoutingTable indexShard : indexRouting) { - set.add(indexShard.shardsRandomIt()); + set.add(preferenceShardIterator(indexShard, clusterState.nodes().localNodeId(), preference)); } } return new GroupShardsIterator(set); } } + private ShardIterator preferenceShardIterator(IndexShardRoutingTable indexShard, String nodeId, @Nullable String preference) { + if (preference == null) { + return indexShard.shardsRandomIt(); + } + if ("_local".equals(preference)) { + return indexShard.preferLocalShardsIt(nodeId); + } + if ("_primary".equals(preference)) { + return indexShard.primaryShardIt(); + } + // if not, then use it as the index + return indexShard.shardsIt(DjbHashFunction.DJB_HASH(preference)); + } + public IndexMetaData indexMetaData(ClusterState clusterState, String index) { IndexMetaData indexMetaData = clusterState.metaData().index(index); if (indexMetaData == null) { diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/get/RestGetAction.java b/modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/get/RestGetAction.java index a2767bc0c3e..cd6d94c5563 100644 --- a/modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/get/RestGetAction.java +++ b/modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/get/RestGetAction.java @@ -59,7 +59,7 @@ public class RestGetAction extends BaseRestHandler { getRequest.operationThreaded(true); getRequest.refresh(request.paramAsBoolean("refresh", getRequest.refresh())); getRequest.routing(request.param("routing")); - + getRequest.preference(request.param("preference")); String sField = request.param("fields"); if (sField != null) { diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java b/modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java index 6b5a80f9081..fdfe1068ff1 100644 --- a/modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java +++ b/modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java @@ -144,6 +144,7 @@ public class RestSearchAction extends BaseRestHandler { searchRequest.types(RestActions.splitTypes(request.param("type"))); searchRequest.queryHint(request.param("query_hint")); searchRequest.routing(request.param("routing")); + searchRequest.preference(request.param("preference")); return searchRequest; } diff --git a/modules/test/integration/src/test/java/org/elasticsearch/test/integration/search/embedded/ThreeShardsEmbeddedSearchTests.java b/modules/test/integration/src/test/java/org/elasticsearch/test/integration/search/embedded/ThreeShardsEmbeddedSearchTests.java index 9131a19347a..ca82833588e 100644 --- a/modules/test/integration/src/test/java/org/elasticsearch/test/integration/search/embedded/ThreeShardsEmbeddedSearchTests.java +++ b/modules/test/integration/src/test/java/org/elasticsearch/test/integration/search/embedded/ThreeShardsEmbeddedSearchTests.java @@ -114,7 +114,7 @@ public class ThreeShardsEmbeddedSearchTests extends AbstractNodesTests { .from(0).size(60).explain(true).indexBoost("test", 1.0f).indexBoost("test2", 2.0f); List dfsResults = newArrayList(); - for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null)) { + for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null, null)) { for (ShardRouting shardRouting : shardIt) { InternalSearchRequest searchRequest = searchRequest(shardRouting, sourceBuilder, SearchType.QUERY_THEN_FETCH) .scroll(new Scroll(new TimeValue(10, TimeUnit.MINUTES))); @@ -182,7 +182,7 @@ public class ThreeShardsEmbeddedSearchTests extends AbstractNodesTests { .from(0).size(60).explain(true).sort("age", SortOrder.ASC); List dfsResults = newArrayList(); - for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null)) { + for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null, null)) { for (ShardRouting shardRouting : shardIt) { InternalSearchRequest searchRequest = searchRequest(shardRouting, sourceBuilder, SearchType.QUERY_THEN_FETCH) .scroll(new Scroll(new TimeValue(10, TimeUnit.MINUTES))); @@ -276,7 +276,7 @@ public class ThreeShardsEmbeddedSearchTests extends AbstractNodesTests { } Map queryFetchResults = newHashMap(); - for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null)) { + for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null, null)) { for (ShardRouting shardRouting : shardIt) { InternalSearchRequest searchRequest = searchRequest(shardRouting, sourceBuilder, SearchType.QUERY_AND_FETCH) .scroll(new Scroll(new TimeValue(10, TimeUnit.MINUTES))); @@ -328,7 +328,7 @@ public class ThreeShardsEmbeddedSearchTests extends AbstractNodesTests { .facet(FacetBuilders.queryFacet("test1", termQuery("name", "test1"))); Map queryResults = newHashMap(); - for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null)) { + for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null, null)) { for (ShardRouting shardRouting : shardIt) { InternalSearchRequest searchRequest = searchRequest(shardRouting, sourceBuilder, SearchType.QUERY_THEN_FETCH) .scroll(new Scroll(new TimeValue(10, TimeUnit.MINUTES))); diff --git a/modules/test/integration/src/test/java/org/elasticsearch/test/integration/search/embedded/ThreeShardsUnbalancedShardsEmbeddedSearchTests.java b/modules/test/integration/src/test/java/org/elasticsearch/test/integration/search/embedded/ThreeShardsUnbalancedShardsEmbeddedSearchTests.java index e1d9ec65c43..6b5e5702023 100644 --- a/modules/test/integration/src/test/java/org/elasticsearch/test/integration/search/embedded/ThreeShardsUnbalancedShardsEmbeddedSearchTests.java +++ b/modules/test/integration/src/test/java/org/elasticsearch/test/integration/search/embedded/ThreeShardsUnbalancedShardsEmbeddedSearchTests.java @@ -122,7 +122,7 @@ public class ThreeShardsUnbalancedShardsEmbeddedSearchTests extends AbstractNode .from(0).size(60).explain(true); List dfsResults = newArrayList(); - for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null)) { + for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null, null)) { for (ShardRouting shardRouting : shardIt) { InternalSearchRequest searchRequest = searchRequest(shardRouting, sourceBuilder, SearchType.DFS_QUERY_THEN_FETCH) .scroll(new Scroll(new TimeValue(10, TimeUnit.MINUTES))); @@ -189,7 +189,7 @@ public class ThreeShardsUnbalancedShardsEmbeddedSearchTests extends AbstractNode .from(0).size(60).explain(true).sort("age", SortOrder.ASC); List dfsResults = newArrayList(); - for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null)) { + for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null, null)) { for (ShardRouting shardRouting : shardIt) { InternalSearchRequest searchRequest = searchRequest(shardRouting, sourceBuilder, SearchType.DFS_QUERY_THEN_FETCH) .scroll(new Scroll(new TimeValue(10, TimeUnit.MINUTES))); @@ -279,7 +279,7 @@ public class ThreeShardsUnbalancedShardsEmbeddedSearchTests extends AbstractNode // do this with dfs, since we have uneven distribution of docs between shards List dfsResults = newArrayList(); - for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null)) { + for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null, null)) { for (ShardRouting shardRouting : shardIt) { InternalSearchRequest searchRequest = searchRequest(shardRouting, sourceBuilder, SearchType.QUERY_AND_FETCH) .scroll(new Scroll(new TimeValue(10, TimeUnit.MINUTES))); @@ -334,7 +334,7 @@ public class ThreeShardsUnbalancedShardsEmbeddedSearchTests extends AbstractNode .facet(queryFacet("test1").query(termQuery("name", "test1"))); Map queryResults = newHashMap(); - for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null)) { + for (ShardIterator shardIt : clusterService.operationRouting().searchShards(clusterService.state(), new String[]{"test"}, null, null, null)) { for (ShardRouting shardRouting : shardIt) { InternalSearchRequest searchRequest = searchRequest(shardRouting, sourceBuilder, SearchType.QUERY_THEN_FETCH) .scroll(new Scroll(new TimeValue(10, TimeUnit.MINUTES))); diff --git a/modules/test/integration/src/test/java/org/elasticsearch/test/integration/search/preference/SearchPreferenceTests.java b/modules/test/integration/src/test/java/org/elasticsearch/test/integration/search/preference/SearchPreferenceTests.java new file mode 100644 index 00000000000..b5019422fab --- /dev/null +++ b/modules/test/integration/src/test/java/org/elasticsearch/test/integration/search/preference/SearchPreferenceTests.java @@ -0,0 +1,77 @@ +/* + * Licensed to Elastic Search and Shay Banon under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. Elastic Search 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.test.integration.search.preference; + +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.client.Client; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.test.integration.AbstractNodesTests; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import static org.elasticsearch.common.settings.ImmutableSettings.*; +import static org.elasticsearch.index.query.xcontent.QueryBuilders.*; +import static org.hamcrest.MatcherAssert.*; +import static org.hamcrest.Matchers.*; + +@Test +public class SearchPreferenceTests extends AbstractNodesTests { + + private Client client; + + @BeforeClass public void createNodes() throws Exception { + Settings settings = settingsBuilder().put("number_of_shards", 3).put("number_of_replicas", 1).build(); + startNode("server1", settings); + startNode("server2", settings); + client = getClient(); + } + + @AfterClass public void closeNodes() { + client.close(); + closeAllNodes(); + } + + protected Client getClient() { + return client("server1"); + } + + @Test public void simplePreferenceTests() throws Exception { + try { + client.admin().indices().prepareDelete("test").execute().actionGet(); + } catch (Exception e) { + // ignore + } + client.admin().indices().prepareCreate("test").execute().actionGet(); + client.admin().cluster().prepareHealth().setWaitForGreenStatus().execute().actionGet(); + + client.prepareIndex("test", "type1").setSource("field1", "value1").execute().actionGet(); + client.admin().indices().prepareRefresh().execute().actionGet(); + + SearchResponse searchResponse = client.prepareSearch().setQuery(matchAllQuery()).setPreference("_local").execute().actionGet(); + assertThat(searchResponse.hits().totalHits(), equalTo(1l)); + + searchResponse = client.prepareSearch().setQuery(matchAllQuery()).setPreference("_primary").execute().actionGet(); + assertThat(searchResponse.hits().totalHits(), equalTo(1l)); + + searchResponse = client.prepareSearch().setQuery(matchAllQuery()).setPreference("1234").execute().actionGet(); + assertThat(searchResponse.hits().totalHits(), equalTo(1l)); + } +} \ No newline at end of file