Shard Allocation: Add a setting to control when rebalancing will happen based on the cluster wide active shards state, closes #814.

modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/RoutingNode.java

@@ -93,6 +93,21 @@ public class RoutingNode implements Iterable<MutableShardRouting> {
         return shards;
     }
 
+    public List<MutableShardRouting> shardsWithState(String index, ShardRoutingState... states) {
+        List<MutableShardRouting> shards = newArrayList();
+        for (MutableShardRouting shardEntry : this) {
+            if (!shardEntry.index().equals(index)) {
+                continue;
+            }
+            for (ShardRoutingState state : states) {
+                if (shardEntry.state() == state) {
+                    shards.add(shardEntry);
+                }
+            }
+        }
+        return shards;
+    }
+
     public int numberOfShardsNotWithState(ShardRoutingState state) {
         int count = 0;
         for (MutableShardRouting shardEntry : this) {

modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/RoutingNodes.java

@@ -193,6 +193,14 @@ public class RoutingNodes implements Iterable<RoutingNode> {
         return shards;
     }
 
+    public List<MutableShardRouting> shardsWithState(String index, ShardRoutingState... state) {
+        List<MutableShardRouting> shards = newArrayList();
+        for (RoutingNode routingNode : this) {
+            shards.addAll(routingNode.shardsWithState(index, state));
+        }
+        return shards;
+    }
+
     public List<RoutingNode> sortedNodesLeastToHigh() {
         return nodesToShardsSorted(new Comparator<RoutingNode>() {
             @Override public int compare(RoutingNode o1, RoutingNode o2) {

modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/allocation/ClusterRebalanceNodeAllocation.java

@@ -0,0 +1,85 @@
+/*
+ * 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.cluster.routing.allocation;
+
+import org.elasticsearch.cluster.routing.MutableShardRouting;
+import org.elasticsearch.cluster.routing.RoutingNode;
+import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+
+public class ClusterRebalanceNodeAllocation extends NodeAllocation {
+
+    public static enum ClusterRebalanceType {
+        ALWAYS,
+        INDICES_PRIMARIES_ACTIVE,
+        INDICES_ALL_ACTIVE
+    }
+
+    private final ClusterRebalanceType type;
+
+    @Inject public ClusterRebalanceNodeAllocation(Settings settings) {
+        super(settings);
+        String allowRebalance = componentSettings.get("allow_rebalance", "indices_all_active");
+        if ("always".equalsIgnoreCase(allowRebalance)) {
+            type = ClusterRebalanceType.ALWAYS;
+        } else if ("indices_primaries_active".equalsIgnoreCase(allowRebalance) || "indicesPrimariesActive".equalsIgnoreCase(allowRebalance)) {
+            type = ClusterRebalanceType.INDICES_PRIMARIES_ACTIVE;
+        } else if ("indices_all_active".equalsIgnoreCase(allowRebalance) || "indicesAllActive".equalsIgnoreCase(allowRebalance)) {
+            type = ClusterRebalanceType.INDICES_ALL_ACTIVE;
+        } else {
+            logger.warn("[cluster.routing.allocation.allow_rebalance] has a wrong value {}, defaulting to 'indices_all_active'", allowRebalance);
+            type = ClusterRebalanceType.INDICES_ALL_ACTIVE;
+        }
+        logger.debug("using [allow_rebalance] with [{}]", type.toString().toLowerCase());
+    }
+
+    @Override public boolean canRebalance(ShardRouting shardRouting, RoutingAllocation allocation) {
+        if (type == ClusterRebalanceType.INDICES_PRIMARIES_ACTIVE) {
+            for (MutableShardRouting shard : allocation.routingNodes().unassigned()) {
+                if (shard.primary()) {
+                    return false;
+                }
+            }
+            for (RoutingNode node : allocation.routingNodes()) {
+                for (MutableShardRouting shard : node) {
+                    if (shard.primary() && !shard.active()) {
+                        return false;
+                    }
+                }
+            }
+            return true;
+        }
+        if (type == ClusterRebalanceType.INDICES_ALL_ACTIVE) {
+            if (!allocation.routingNodes().unassigned().isEmpty()) {
+                return false;
+            }
+            for (RoutingNode node : allocation.routingNodes()) {
+                for (MutableShardRouting shard : node) {
+                    if (!shard.active()) {
+                        return false;
+                    }
+                }
+            }
+        }
+        // type == Type.ALWAYS
+        return true;
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/allocation/NodeAllocations.java

@@ -42,6 +42,7 @@ public class NodeAllocations extends NodeAllocation {
                 .add(new ReplicaAfterPrimaryActiveNodeAllocation(settings))
                 .add(new ThrottlingNodeAllocation(settings))
                 .add(new RebalanceOnlyWhenActiveNodeAllocation(settings))
+                .add(new ClusterRebalanceNodeAllocation(settings))
                 .build()
         );
     }

modules/elasticsearch/src/main/java/org/elasticsearch/cluster/routing/allocation/ShardAllocationModule.java

@@ -48,6 +48,7 @@ public class ShardAllocationModule extends AbstractModule {
         allocationMultibinder.addBinding().to(ReplicaAfterPrimaryActiveNodeAllocation.class);
         allocationMultibinder.addBinding().to(ThrottlingNodeAllocation.class);
         allocationMultibinder.addBinding().to(RebalanceOnlyWhenActiveNodeAllocation.class);
+        allocationMultibinder.addBinding().to(ClusterRebalanceNodeAllocation.class);
         for (Class<? extends NodeAllocation> allocation : allocations) {
             allocationMultibinder.addBinding().to(allocation);
         }

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/ClusterRebalanceRoutingTests.java

@@ -0,0 +1,623 @@
+/*
+ * 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.cluster.routing.allocation;
+
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.cluster.routing.RoutingNodes;
+import org.elasticsearch.cluster.routing.RoutingTable;
+import org.elasticsearch.common.logging.ESLogger;
+import org.elasticsearch.common.logging.Loggers;
+import org.testng.annotations.Test;
+
+import static org.elasticsearch.cluster.ClusterState.*;
+import static org.elasticsearch.cluster.metadata.IndexMetaData.*;
+import static org.elasticsearch.cluster.metadata.MetaData.*;
+import static org.elasticsearch.cluster.node.DiscoveryNodes.*;
+import static org.elasticsearch.cluster.routing.RoutingBuilders.*;
+import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
+import static org.elasticsearch.cluster.routing.allocation.RoutingAllocationTests.*;
+import static org.elasticsearch.common.settings.ImmutableSettings.*;
+import static org.hamcrest.MatcherAssert.*;
+import static org.hamcrest.Matchers.*;
+
+@Test
+public class ClusterRebalanceRoutingTests {
+
+    private final ESLogger logger = Loggers.getLogger(ClusterRebalanceRoutingTests.class);
+
+    @Test public void testAlways() {
+        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.allow_rebalance", ClusterRebalanceNodeAllocation.ClusterRebalanceType.ALWAYS.toString()).build());
+
+        MetaData metaData = newMetaDataBuilder()
+                .put(newIndexMetaDataBuilder("test1").numberOfShards(1).numberOfReplicas(1))
+                .put(newIndexMetaDataBuilder("test2").numberOfShards(1).numberOfReplicas(1))
+                .build();
+
+        RoutingTable routingTable = routingTable()
+                .add(indexRoutingTable("test1").initializeEmpty(metaData.index("test1")))
+                .add(indexRoutingTable("test2").initializeEmpty(metaData.index("test2")))
+                .build();
+
+        ClusterState clusterState = newClusterStateBuilder().metaData(metaData).routingTable(routingTable).build();
+
+        logger.info("start two nodes");
+        clusterState = newClusterStateBuilder().state(clusterState).nodes(newNodesBuilder().put(newNode("node1")).put(newNode("node2"))).build();
+        RoutingTable prevRoutingTable = routingTable;
+        routingTable = strategy.reroute(clusterState).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start all the primary shards for test1, replicas will start initializing");
+        RoutingNodes routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test1", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(INITIALIZING));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start the test1 replica shards");
+        routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test1", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(STARTED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("now, start 1 more node, check that rebalancing will happen (for test1) because we set it to always");
+        clusterState = newClusterStateBuilder().state(clusterState).nodes(newNodesBuilder().putAll(clusterState.nodes())
+                .put(newNode("node3")))
+                .build();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.reroute(clusterState).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        assertThat(routingNodes.node("node3").shards().size(), equalTo(1));
+        assertThat(routingNodes.node("node3").shards().get(0).shardId().index().name(), equalTo("test1"));
+    }
+
+
+    @Test public void testClusterPrimariesActive1() {
+        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.allow_rebalance", ClusterRebalanceNodeAllocation.ClusterRebalanceType.INDICES_PRIMARIES_ACTIVE.toString()).build());
+
+        MetaData metaData = newMetaDataBuilder()
+                .put(newIndexMetaDataBuilder("test1").numberOfShards(1).numberOfReplicas(1))
+                .put(newIndexMetaDataBuilder("test2").numberOfShards(1).numberOfReplicas(1))
+                .build();
+
+        RoutingTable routingTable = routingTable()
+                .add(indexRoutingTable("test1").initializeEmpty(metaData.index("test1")))
+                .add(indexRoutingTable("test2").initializeEmpty(metaData.index("test2")))
+                .build();
+
+        ClusterState clusterState = newClusterStateBuilder().metaData(metaData).routingTable(routingTable).build();
+
+        logger.info("start two nodes");
+        clusterState = newClusterStateBuilder().state(clusterState).nodes(newNodesBuilder().put(newNode("node1")).put(newNode("node2"))).build();
+        RoutingTable prevRoutingTable = routingTable;
+        routingTable = strategy.reroute(clusterState).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start all the primary shards for test1, replicas will start initializing");
+        RoutingNodes routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test1", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(INITIALIZING));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start the test1 replica shards");
+        routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test1", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(STARTED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start all the primary shards for test2, replicas will start initializing");
+        routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test2", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(STARTED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(INITIALIZING));
+        }
+
+        logger.info("now, start 1 more node, check that rebalancing happen (for test1) because we set it to primaries_active");
+        clusterState = newClusterStateBuilder().state(clusterState).nodes(newNodesBuilder().putAll(clusterState.nodes())
+                .put(newNode("node3")))
+                .build();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.reroute(clusterState).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        assertThat(routingNodes.node("node3").shards().size(), equalTo(1));
+        assertThat(routingNodes.node("node3").shards().get(0).shardId().index().name(), equalTo("test1"));
+    }
+
+    @Test public void testClusterPrimariesActive2() {
+        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.allow_rebalance", ClusterRebalanceNodeAllocation.ClusterRebalanceType.INDICES_PRIMARIES_ACTIVE.toString()).build());
+
+        MetaData metaData = newMetaDataBuilder()
+                .put(newIndexMetaDataBuilder("test1").numberOfShards(1).numberOfReplicas(1))
+                .put(newIndexMetaDataBuilder("test2").numberOfShards(1).numberOfReplicas(1))
+                .build();
+
+        RoutingTable routingTable = routingTable()
+                .add(indexRoutingTable("test1").initializeEmpty(metaData.index("test1")))
+                .add(indexRoutingTable("test2").initializeEmpty(metaData.index("test2")))
+                .build();
+
+        ClusterState clusterState = newClusterStateBuilder().metaData(metaData).routingTable(routingTable).build();
+
+        logger.info("start two nodes");
+        clusterState = newClusterStateBuilder().state(clusterState).nodes(newNodesBuilder().put(newNode("node1")).put(newNode("node2"))).build();
+        RoutingTable prevRoutingTable = routingTable;
+        routingTable = strategy.reroute(clusterState).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start all the primary shards for test1, replicas will start initializing");
+        RoutingNodes routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test1", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(INITIALIZING));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start the test1 replica shards");
+        routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test1", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(STARTED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("now, start 1 more node, check that rebalancing will not happen (for test1) because we set it to primaries_active");
+        clusterState = newClusterStateBuilder().state(clusterState).nodes(newNodesBuilder().putAll(clusterState.nodes())
+                .put(newNode("node3")))
+                .build();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.reroute(clusterState).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        assertThat(routingNodes.node("node3"), nullValue());
+    }
+
+    @Test public void testClusterAllActive1() {
+        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.allow_rebalance", ClusterRebalanceNodeAllocation.ClusterRebalanceType.INDICES_ALL_ACTIVE.toString()).build());
+
+        MetaData metaData = newMetaDataBuilder()
+                .put(newIndexMetaDataBuilder("test1").numberOfShards(1).numberOfReplicas(1))
+                .put(newIndexMetaDataBuilder("test2").numberOfShards(1).numberOfReplicas(1))
+                .build();
+
+        RoutingTable routingTable = routingTable()
+                .add(indexRoutingTable("test1").initializeEmpty(metaData.index("test1")))
+                .add(indexRoutingTable("test2").initializeEmpty(metaData.index("test2")))
+                .build();
+
+        ClusterState clusterState = newClusterStateBuilder().metaData(metaData).routingTable(routingTable).build();
+
+        logger.info("start two nodes");
+        clusterState = newClusterStateBuilder().state(clusterState).nodes(newNodesBuilder().put(newNode("node1")).put(newNode("node2"))).build();
+        RoutingTable prevRoutingTable = routingTable;
+        routingTable = strategy.reroute(clusterState).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start all the primary shards for test1, replicas will start initializing");
+        RoutingNodes routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test1", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(INITIALIZING));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start the test1 replica shards");
+        routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test1", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(STARTED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start all the primary shards for test2, replicas will start initializing");
+        routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test2", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(STARTED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(INITIALIZING));
+        }
+
+        logger.info("start the test2 replica shards");
+        routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test2", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(STARTED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(STARTED));
+        }
+
+        logger.info("now, start 1 more node, check that rebalancing happen (for test1) because we set it to all_active");
+        clusterState = newClusterStateBuilder().state(clusterState).nodes(newNodesBuilder().putAll(clusterState.nodes())
+                .put(newNode("node3")))
+                .build();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.reroute(clusterState).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        assertThat(routingNodes.node("node3").shards().size(), equalTo(1));
+        assertThat(routingNodes.node("node3").shards().get(0).shardId().index().name(), equalTo("test1"));
+    }
+
+    @Test public void testClusterAllActive2() {
+        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.allow_rebalance", ClusterRebalanceNodeAllocation.ClusterRebalanceType.INDICES_ALL_ACTIVE.toString()).build());
+
+        MetaData metaData = newMetaDataBuilder()
+                .put(newIndexMetaDataBuilder("test1").numberOfShards(1).numberOfReplicas(1))
+                .put(newIndexMetaDataBuilder("test2").numberOfShards(1).numberOfReplicas(1))
+                .build();
+
+        RoutingTable routingTable = routingTable()
+                .add(indexRoutingTable("test1").initializeEmpty(metaData.index("test1")))
+                .add(indexRoutingTable("test2").initializeEmpty(metaData.index("test2")))
+                .build();
+
+        ClusterState clusterState = newClusterStateBuilder().metaData(metaData).routingTable(routingTable).build();
+
+        logger.info("start two nodes");
+        clusterState = newClusterStateBuilder().state(clusterState).nodes(newNodesBuilder().put(newNode("node1")).put(newNode("node2"))).build();
+        RoutingTable prevRoutingTable = routingTable;
+        routingTable = strategy.reroute(clusterState).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start all the primary shards for test1, replicas will start initializing");
+        RoutingNodes routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test1", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(INITIALIZING));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start the test1 replica shards");
+        routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test1", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(STARTED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("now, start 1 more node, check that rebalancing will not happen (for test1) because we set it to all_active");
+        clusterState = newClusterStateBuilder().state(clusterState).nodes(newNodesBuilder().putAll(clusterState.nodes())
+                .put(newNode("node3")))
+                .build();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.reroute(clusterState).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        assertThat(routingNodes.node("node3"), nullValue());
+    }
+
+    @Test public void testClusterAllActive3() {
+        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.allow_rebalance", ClusterRebalanceNodeAllocation.ClusterRebalanceType.INDICES_ALL_ACTIVE.toString()).build());
+
+        MetaData metaData = newMetaDataBuilder()
+                .put(newIndexMetaDataBuilder("test1").numberOfShards(1).numberOfReplicas(1))
+                .put(newIndexMetaDataBuilder("test2").numberOfShards(1).numberOfReplicas(1))
+                .build();
+
+        RoutingTable routingTable = routingTable()
+                .add(indexRoutingTable("test1").initializeEmpty(metaData.index("test1")))
+                .add(indexRoutingTable("test2").initializeEmpty(metaData.index("test2")))
+                .build();
+
+        ClusterState clusterState = newClusterStateBuilder().metaData(metaData).routingTable(routingTable).build();
+
+        logger.info("start two nodes");
+        clusterState = newClusterStateBuilder().state(clusterState).nodes(newNodesBuilder().put(newNode("node1")).put(newNode("node2"))).build();
+        RoutingTable prevRoutingTable = routingTable;
+        routingTable = strategy.reroute(clusterState).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start all the primary shards for test1, replicas will start initializing");
+        RoutingNodes routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test1", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(INITIALIZING));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start the test1 replica shards");
+        routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test1", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(STARTED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(INITIALIZING));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(UNASSIGNED));
+        }
+
+        logger.info("start all the primary shards for test2, replicas will start initializing");
+        routingNodes = clusterState.routingNodes();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.applyStartedShards(clusterState, routingNodes.shardsWithState("test2", INITIALIZING)).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        for (int i = 0; i < routingTable.index("test1").shards().size(); i++) {
+            assertThat(routingTable.index("test1").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test1").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test1").shard(i).replicaShards().get(0).state(), equalTo(STARTED));
+        }
+
+        for (int i = 0; i < routingTable.index("test2").shards().size(); i++) {
+            assertThat(routingTable.index("test2").shard(i).shards().size(), equalTo(2));
+            assertThat(routingTable.index("test2").shard(i).primaryShard().state(), equalTo(STARTED));
+            assertThat(routingTable.index("test2").shard(i).replicaShards().get(0).state(), equalTo(INITIALIZING));
+        }
+
+        logger.info("now, start 1 more node, check that rebalancing will not happen (for test1) because we set it to all_active");
+        clusterState = newClusterStateBuilder().state(clusterState).nodes(newNodesBuilder().putAll(clusterState.nodes())
+                .put(newNode("node3")))
+                .build();
+        prevRoutingTable = routingTable;
+        routingTable = strategy.reroute(clusterState).routingTable();
+        clusterState = newClusterStateBuilder().state(clusterState).routingTable(routingTable).build();
+        routingNodes = clusterState.routingNodes();
+
+        assertThat(routingNodes.node("node3"), nullValue());
+    }
+}

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/ElectReplicaAsPrimaryDuringRelocationTests.java

@@ -21,13 +21,11 @@ package org.elasticsearch.cluster.routing.allocation;
 
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
-import org.elasticsearch.common.transport.DummyTransportAddress;
 import org.testng.annotations.Test;
 
 import static org.elasticsearch.cluster.ClusterState.*;
@@ -36,6 +34,7 @@ import static org.elasticsearch.cluster.metadata.MetaData.*;
 import static org.elasticsearch.cluster.node.DiscoveryNodes.*;
 import static org.elasticsearch.cluster.routing.RoutingBuilders.*;
 import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
+import static org.elasticsearch.cluster.routing.allocation.RoutingAllocationTests.*;
 import static org.elasticsearch.common.settings.ImmutableSettings.*;
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
@@ -112,8 +111,4 @@ public class ElectReplicaAsPrimaryDuringRelocationTests {
         assertThat(routingTable.index("test").shard(0).primaryShard().active(), equalTo(true));
         assertThat(routingTable.index("test").shard(1).primaryShard().active(), equalTo(true));
     }
-
-    private DiscoveryNode newNode(String nodeId) {
-        return new DiscoveryNode(nodeId, DummyTransportAddress.INSTANCE);
-    }
 }

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/FailedShardsRoutingTests.java

@@ -21,13 +21,11 @@ package org.elasticsearch.cluster.routing.allocation;
 
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.MutableShardRouting;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
-import org.elasticsearch.common.transport.DummyTransportAddress;
 import org.testng.annotations.Test;
 
 import java.util.List;
@@ -38,6 +36,7 @@ import static org.elasticsearch.cluster.metadata.MetaData.*;
 import static org.elasticsearch.cluster.node.DiscoveryNodes.*;
 import static org.elasticsearch.cluster.routing.RoutingBuilders.*;
 import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
+import static org.elasticsearch.cluster.routing.allocation.RoutingAllocationTests.*;
 import static org.elasticsearch.common.settings.ImmutableSettings.*;
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
@@ -51,7 +50,10 @@ public class FailedShardsRoutingTests {
     private final ESLogger logger = Loggers.getLogger(FailedShardsRoutingTests.class);
 
     @Test public void testFailures() {
-        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.concurrent_recoveries", 10).build());
+        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder()
+                .put("cluster.routing.allocation.concurrent_recoveries", 10)
+                .put("cluster.routing.allocation.allow_rebalance", "always")
+                .build());
 
         logger.info("Building initial routing table");
 
@@ -287,8 +289,4 @@ public class FailedShardsRoutingTests {
             assertThat(routingTable.index("test").shard(i).replicaShards().get(0).currentNodeId(), nullValue());
         }
     }
-
-    private DiscoveryNode newNode(String nodeId) {
-        return new DiscoveryNode(nodeId, DummyTransportAddress.INSTANCE);
-    }
 }

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/PrimaryElectionRoutingTests.java

@@ -21,12 +21,10 @@ package org.elasticsearch.cluster.routing.allocation;
 
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
-import org.elasticsearch.common.transport.DummyTransportAddress;
 import org.testng.annotations.Test;
 
 import static org.elasticsearch.cluster.ClusterState.*;
@@ -35,6 +33,7 @@ import static org.elasticsearch.cluster.metadata.MetaData.*;
 import static org.elasticsearch.cluster.node.DiscoveryNodes.*;
 import static org.elasticsearch.cluster.routing.RoutingBuilders.*;
 import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
+import static org.elasticsearch.cluster.routing.allocation.RoutingAllocationTests.*;
 import static org.elasticsearch.common.settings.ImmutableSettings.*;
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
@@ -95,8 +94,4 @@ public class PrimaryElectionRoutingTests {
         assertThat(routingTable.index("test").shard(0).primaryShard().currentNodeId(), equalTo("node2"));
         assertThat(routingTable.index("test").shard(0).replicaShards().get(0).currentNodeId(), equalTo("node3"));
     }
-
-    private DiscoveryNode newNode(String nodeId) {
-        return new DiscoveryNode(nodeId, DummyTransportAddress.INSTANCE);
-    }
 }

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/PrimaryNotRelocatedWhileBeingRecoveredTests.java

@@ -21,12 +21,10 @@ package org.elasticsearch.cluster.routing.allocation;
 
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
-import org.elasticsearch.common.transport.DummyTransportAddress;
 import org.testng.annotations.Test;
 
 import static org.elasticsearch.cluster.ClusterState.*;
@@ -35,6 +33,7 @@ import static org.elasticsearch.cluster.metadata.MetaData.*;
 import static org.elasticsearch.cluster.node.DiscoveryNodes.*;
 import static org.elasticsearch.cluster.routing.RoutingBuilders.*;
 import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
+import static org.elasticsearch.cluster.routing.allocation.RoutingAllocationTests.*;
 import static org.elasticsearch.common.settings.ImmutableSettings.*;
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
@@ -90,8 +89,4 @@ public class PrimaryNotRelocatedWhileBeingRecoveredTests {
         assertThat(routingTable.shardsWithState(STARTED).size(), equalTo(5));
         assertThat(routingTable.shardsWithState(INITIALIZING).size(), equalTo(5));
     }
-
-    private DiscoveryNode newNode(String nodeId) {
-        return new DiscoveryNode(nodeId, DummyTransportAddress.INSTANCE);
-    }
 }

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/RebalanceAfterActiveTests.java

@@ -21,13 +21,11 @@ package org.elasticsearch.cluster.routing.allocation;
 
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingNode;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
-import org.elasticsearch.common.transport.DummyTransportAddress;
 import org.testng.annotations.Test;
 
 import static org.elasticsearch.cluster.ClusterState.*;
@@ -36,6 +34,7 @@ import static org.elasticsearch.cluster.metadata.MetaData.*;
 import static org.elasticsearch.cluster.node.DiscoveryNodes.*;
 import static org.elasticsearch.cluster.routing.RoutingBuilders.*;
 import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
+import static org.elasticsearch.cluster.routing.allocation.RoutingAllocationTests.*;
 import static org.elasticsearch.common.settings.ImmutableSettings.*;
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
@@ -48,7 +47,10 @@ public class RebalanceAfterActiveTests {
     private final ESLogger logger = Loggers.getLogger(RebalanceAfterActiveTests.class);
 
     @Test public void testRebalanceOnlyAfterAllShardsAreActive() {
-        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.concurrent_recoveries", 10).build());
+        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder()
+                .put("cluster.routing.allocation.concurrent_recoveries", 10)
+                .put("cluster.routing.allocation.allow_rebalance", "always")
+                .build());
 
         logger.info("Building initial routing table");
 
@@ -146,8 +148,4 @@ public class RebalanceAfterActiveTests {
             assertThat(routingNode.shards().size(), equalTo(1));
         }
     }
-
-    private DiscoveryNode newNode(String nodeId) {
-        return new DiscoveryNode(nodeId, DummyTransportAddress.INSTANCE);
-    }
 }

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/ReplicaAllocatedAfterPrimaryTests.java

@@ -21,12 +21,10 @@ package org.elasticsearch.cluster.routing.allocation;
 
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
-import org.elasticsearch.common.transport.DummyTransportAddress;
 import org.testng.annotations.Test;
 
 import static org.elasticsearch.cluster.ClusterState.*;
@@ -35,6 +33,7 @@ import static org.elasticsearch.cluster.metadata.MetaData.*;
 import static org.elasticsearch.cluster.node.DiscoveryNodes.*;
 import static org.elasticsearch.cluster.routing.RoutingBuilders.*;
 import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
+import static org.elasticsearch.cluster.routing.allocation.RoutingAllocationTests.*;
 import static org.elasticsearch.common.settings.ImmutableSettings.*;
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
@@ -103,8 +102,4 @@ public class ReplicaAllocatedAfterPrimaryTests {
         assertThat(routingTable.index("test").shard(0).replicaShards().get(0).currentNodeId(), equalTo("node2"));
 
     }
-
-    private DiscoveryNode newNode(String nodeId) {
-        return new DiscoveryNode(nodeId, DummyTransportAddress.INSTANCE);
-    }
 }

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/RoutingAllocationTests.java

@@ -0,0 +1,30 @@
+/*
+ * 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.cluster.routing.allocation;
+
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.transport.DummyTransportAddress;
+
+public class RoutingAllocationTests {
+
+    public static DiscoveryNode newNode(String nodeId) {
+        return new DiscoveryNode(nodeId, DummyTransportAddress.INSTANCE);
+    }
+}

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/SingleShardNoReplicasRoutingTests.java

@@ -29,7 +29,6 @@ import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
-import org.elasticsearch.common.transport.DummyTransportAddress;
 import org.testng.annotations.Test;
 
 import java.util.List;
@@ -41,6 +40,7 @@ import static org.elasticsearch.cluster.metadata.MetaData.*;
 import static org.elasticsearch.cluster.node.DiscoveryNodes.*;
 import static org.elasticsearch.cluster.routing.RoutingBuilders.*;
 import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
+import static org.elasticsearch.cluster.routing.allocation.RoutingAllocationTests.*;
 import static org.elasticsearch.common.collect.Lists.*;
 import static org.elasticsearch.common.collect.Sets.*;
 import static org.elasticsearch.common.settings.ImmutableSettings.*;
@@ -205,7 +205,10 @@ public class SingleShardNoReplicasRoutingTests {
     }
 
     @Test public void testMultiIndexEvenDistribution() {
-        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.concurrent_recoveries", 10).build());
+        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder()
+                .put("cluster.routing.allocation.concurrent_recoveries", 10)
+                .put("cluster.routing.allocation.allow_rebalance", "always")
+                .build());
 
         final int numberOfIndices = 50;
         logger.info("Building initial routing table with " + numberOfIndices + " indices");
@@ -313,7 +316,10 @@ public class SingleShardNoReplicasRoutingTests {
     }
 
     @Test public void testMultiIndexUnevenNodes() {
-        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.concurrent_recoveries", 10).build());
+        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder()
+                .put("cluster.routing.allocation.concurrent_recoveries", 10)
+                .put("cluster.routing.allocation.allow_rebalance", "always")
+                .build());
 
         final int numberOfIndices = 10;
         logger.info("Building initial routing table with " + numberOfIndices + " indices");
@@ -401,8 +407,4 @@ public class SingleShardNoReplicasRoutingTests {
             assertThat(routingNode.numberOfShardsWithState(STARTED), equalTo(2));
         }
     }
-
-    private DiscoveryNode newNode(String nodeId) {
-        return new DiscoveryNode(nodeId, DummyTransportAddress.INSTANCE);
-    }
 }

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/SingleShardOneReplicaRoutingTests.java

@@ -21,12 +21,10 @@ package org.elasticsearch.cluster.routing.allocation;
 
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
-import org.elasticsearch.common.transport.DummyTransportAddress;
 import org.testng.annotations.Test;
 
 import static org.elasticsearch.cluster.ClusterState.*;
@@ -35,6 +33,7 @@ import static org.elasticsearch.cluster.metadata.MetaData.*;
 import static org.elasticsearch.cluster.node.DiscoveryNodes.*;
 import static org.elasticsearch.cluster.routing.RoutingBuilders.*;
 import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
+import static org.elasticsearch.cluster.routing.allocation.RoutingAllocationTests.*;
 import static org.elasticsearch.common.settings.ImmutableSettings.*;
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
@@ -169,8 +168,4 @@ public class SingleShardOneReplicaRoutingTests {
         assertThat(routingTable.index("test").shard(0).replicaShards().get(0).state(), equalTo(INITIALIZING));
         assertThat(routingTable.index("test").shard(0).replicaShards().get(0).currentNodeId(), equalTo("node3"));
     }
-
-    private DiscoveryNode newNode(String nodeId) {
-        return new DiscoveryNode(nodeId, DummyTransportAddress.INSTANCE);
-    }
 }

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/TenShardsOneReplicaRoutingTests.java

@@ -21,12 +21,10 @@ package org.elasticsearch.cluster.routing.allocation;
 
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
-import org.elasticsearch.common.transport.DummyTransportAddress;
 import org.testng.annotations.Test;
 
 import static org.elasticsearch.cluster.ClusterState.*;
@@ -35,6 +33,7 @@ import static org.elasticsearch.cluster.metadata.MetaData.*;
 import static org.elasticsearch.cluster.node.DiscoveryNodes.*;
 import static org.elasticsearch.cluster.routing.RoutingBuilders.*;
 import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
+import static org.elasticsearch.cluster.routing.allocation.RoutingAllocationTests.*;
 import static org.elasticsearch.common.settings.ImmutableSettings.*;
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
@@ -47,7 +46,10 @@ public class TenShardsOneReplicaRoutingTests {
     private final ESLogger logger = Loggers.getLogger(TenShardsOneReplicaRoutingTests.class);
 
     @Test public void testSingleIndexFirstStartPrimaryThenBackups() {
-        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.concurrent_recoveries", 10).build());
+        ShardsAllocation strategy = new ShardsAllocation(settingsBuilder()
+                .put("cluster.routing.allocation.concurrent_recoveries", 10)
+                .put("cluster.routing.allocation.allow_rebalance", "always")
+                .build());
 
         logger.info("Building initial routing table");
 
@@ -171,8 +173,4 @@ public class TenShardsOneReplicaRoutingTests {
         assertThat(routingNodes.node("node2").numberOfShardsWithState(STARTED), equalTo(6));
         assertThat(routingNodes.node("node3").numberOfShardsWithState(STARTED), equalTo(6));
     }
-
-    private DiscoveryNode newNode(String nodeId) {
-        return new DiscoveryNode(nodeId, DummyTransportAddress.INSTANCE);
-    }
 }

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/ThrottlingAllocationTests.java

@@ -21,11 +21,9 @@ package org.elasticsearch.cluster.routing.allocation;
 
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
-import org.elasticsearch.common.transport.DummyTransportAddress;
 import org.testng.annotations.Test;
 
 import static org.elasticsearch.cluster.ClusterState.*;
@@ -34,6 +32,7 @@ import static org.elasticsearch.cluster.metadata.MetaData.*;
 import static org.elasticsearch.cluster.node.DiscoveryNodes.*;
 import static org.elasticsearch.cluster.routing.RoutingBuilders.*;
 import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
+import static org.elasticsearch.cluster.routing.allocation.RoutingAllocationTests.*;
 import static org.elasticsearch.common.settings.ImmutableSettings.*;
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
@@ -167,8 +166,4 @@ public class ThrottlingAllocationTests {
         assertThat(routingTable.shardsWithState(INITIALIZING).size(), equalTo(0));
         assertThat(routingTable.shardsWithState(UNASSIGNED).size(), equalTo(0));
     }
-
-    private DiscoveryNode newNode(String nodeId) {
-        return new DiscoveryNode(nodeId, DummyTransportAddress.INSTANCE);
-    }
 }

modules/elasticsearch/src/test/java/org/elasticsearch/cluster/routing/allocation/UpdateNumberOfReplicasTests.java

@@ -2,12 +2,10 @@ package org.elasticsearch.cluster.routing.allocation;
 
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.logging.Loggers;
-import org.elasticsearch.common.transport.DummyTransportAddress;
 import org.testng.annotations.Test;
 
 import static org.elasticsearch.cluster.ClusterState.*;
@@ -17,6 +15,7 @@ import static org.elasticsearch.cluster.node.DiscoveryNodes.*;
 import static org.elasticsearch.cluster.routing.RoutingBuilders.*;
 import static org.elasticsearch.cluster.routing.RoutingTable.*;
 import static org.elasticsearch.cluster.routing.ShardRoutingState.*;
+import static org.elasticsearch.cluster.routing.allocation.RoutingAllocationTests.*;
 import static org.elasticsearch.common.settings.ImmutableSettings.*;
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
@@ -158,8 +157,4 @@ public class UpdateNumberOfReplicasTests {
 
         assertThat(prevRoutingTable != routingTable, equalTo(false));
     }
-
-    private DiscoveryNode newNode(String nodeId) {
-        return new DiscoveryNode(nodeId, DummyTransportAddress.INSTANCE);
-    }
 }