Allow ShardsAllocator to be configured via node level settings.

* Default ShardsAllocator is set to BalancedShardsAllocator
* Core ShardsAllocator implementations can be defined via 'cluster.routing.allocation.type'
* Core ShardsAllocator implementations are exposed via short keys 'balanced' (BalancedShardsAllocator) and 'even_shards' (EvenShardsCountAllocator)
* Third party allocators can be loaded via fully-qualified class names.

Closes #2557

src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/ShardsAllocatorModule.java

@@ -26,6 +26,12 @@ import org.elasticsearch.gateway.none.NoneGatewayAllocator;
 /**
  */
 public class ShardsAllocatorModule extends AbstractModule {
+	
+    public static final String EVEN_SHARD_COUNT_ALLOCATOR_KEY = "even_shard";
+
+    public static final String BALANCED_ALLOCATOR_KEY = "balanced"; // default
+
+    public static final String TYPE_KEY = "cluster.routing.allocation.type";
 
     private Settings settings;
 
@@ -35,6 +41,7 @@ public class ShardsAllocatorModule extends AbstractModule {
 
     public ShardsAllocatorModule(Settings settings) {
         this.settings = settings;
+        shardsAllocator = loadShardsAllocator(settings);
     }
 
     public void setGatewayAllocator(Class<? extends GatewayAllocator> gatewayAllocator) {
@@ -47,7 +54,24 @@ public class ShardsAllocatorModule extends AbstractModule {
 
     @Override
     protected void configure() {
+        if (shardsAllocator == null) {
+            shardsAllocator = loadShardsAllocator(settings);
+        }
         bind(GatewayAllocator.class).to(gatewayAllocator).asEagerSingleton();
-        bind(ShardsAllocator.class).to(shardsAllocator == null ? BalancedShardsAllocator.class : shardsAllocator).asEagerSingleton();
+        bind(ShardsAllocator.class).to(shardsAllocator).asEagerSingleton();
+    }
+    
+    private Class<? extends ShardsAllocator> loadShardsAllocator(Settings settings) {
+        final Class<? extends ShardsAllocator> shardsAllocator;
+        final String type = settings.get(TYPE_KEY, BALANCED_ALLOCATOR_KEY);
+        if (BALANCED_ALLOCATOR_KEY.equals(type)) {
+            shardsAllocator = BalancedShardsAllocator.class;
+        } else if (EVEN_SHARD_COUNT_ALLOCATOR_KEY.equals(type)) {
+            shardsAllocator = EvenShardsCountAllocator.class;
+        } else {
+            shardsAllocator = settings.getAsClass(TYPE_KEY, BalancedShardsAllocator.class,
+                    "org.elasticsearch.cluster.routing.allocation.allocator.", "Allocator");
+        }
+        return shardsAllocator;
     }
 }

src/test/java/org/elasticsearch/test/unit/cluster/routing/allocation/ShardsAllocatorModuleTests.java

@@ -0,0 +1,75 @@
+package org.elasticsearch.test.unit.cluster.routing.allocation;
+
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch 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.
+ */
+
+import static org.elasticsearch.common.settings.ImmutableSettings.settingsBuilder;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.instanceOf;
+
+import org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator;
+import org.elasticsearch.cluster.routing.allocation.allocator.EvenShardsCountAllocator;
+import org.elasticsearch.cluster.routing.allocation.allocator.ShardsAllocator;
+import org.elasticsearch.cluster.routing.allocation.allocator.ShardsAllocatorModule;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.node.Node;
+import org.elasticsearch.node.internal.InternalNode;
+import org.elasticsearch.test.integration.AbstractNodesTests;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.Test;
+
+@Test
+public class ShardsAllocatorModuleTests extends AbstractNodesTests {
+
+    @AfterMethod
+    public void cleanAndCloseNodes() throws Exception {
+        closeAllNodes();
+    }
+
+    public void testLoadDefaultShardsAllocator() {
+        assertAllocatorInstance(ImmutableSettings.Builder.EMPTY_SETTINGS, BalancedShardsAllocator.class);
+    }
+
+    public void testLoadByShortKeyShardsAllocator() {
+        Settings build = settingsBuilder().put(ShardsAllocatorModule.TYPE_KEY, ShardsAllocatorModule.EVEN_SHARD_COUNT_ALLOCATOR_KEY)
+                .build();
+        assertAllocatorInstance(build, EvenShardsCountAllocator.class);
+
+        build = settingsBuilder().put(ShardsAllocatorModule.TYPE_KEY, ShardsAllocatorModule.BALANCED_ALLOCATOR_KEY).build();
+        assertAllocatorInstance(build, BalancedShardsAllocator.class);
+    }
+
+    public void testLoadByClassNameShardsAllocator() {
+        Settings build = settingsBuilder().put(ShardsAllocatorModule.TYPE_KEY, "EvenShardsCount").build();
+        assertAllocatorInstance(build, EvenShardsCountAllocator.class);
+
+        build = settingsBuilder().put(ShardsAllocatorModule.TYPE_KEY,
+                "org.elasticsearch.cluster.routing.allocation.allocator.EvenShardsCountAllocator").build();
+        assertAllocatorInstance(build, EvenShardsCountAllocator.class);
+    }
+
+    private void assertAllocatorInstance(Settings settings, Class<? extends ShardsAllocator> clazz) {
+        Node _node = startNode("node", settings);
+        InternalNode node = (InternalNode) _node;
+        ShardsAllocator instance = node.injector().getInstance(ShardsAllocator.class);
+        node.close();
+        assertThat(instance, instanceOf(clazz));
+    }
+}