configurable thread count

server/src/main/java/io/druid/server/coordinator/CostBalancerMultithreadStrategy.java

@@ -36,10 +36,12 @@ import java.util.concurrent.Executors;
 public class CostBalancerMultithreadStrategy extends AbstractCostBalancerStrategy
 {
   private static final EmittingLogger log = new EmittingLogger(CostBalancerMultithreadStrategy.class);
+  private final int threadCount;
 
-  public CostBalancerMultithreadStrategy(DateTime referenceTimestamp)
+  public CostBalancerMultithreadStrategy(DateTime referenceTimestamp, int threadCount)
   {
     super(referenceTimestamp);
+    this.threadCount = threadCount;
   }
 
   protected Pair<Double, ServerHolder> chooseBestServer(
@@ -50,7 +52,7 @@ public class CostBalancerMultithreadStrategy extends AbstractCostBalancerStrateg
   {
     Pair<Double, ServerHolder> bestServer = Pair.of(Double.POSITIVE_INFINITY, null);
 
-    ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(8));
+    ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(threadCount));
     List<ListenableFuture<Pair<Double, ServerHolder>>> futures = Lists.newArrayList();
 
     for (final ServerHolder server : serverHolders) {

server/src/main/java/io/druid/server/coordinator/CostBalancerMultithreadStrategyFactory.java

@@ -22,10 +22,16 @@ import org.joda.time.DateTime;
 
 public class CostBalancerMultithreadStrategyFactory implements BalancerStrategyFactory
 {
+  private final int threadCount;
+
+  public CostBalancerMultithreadStrategyFactory(int costBalancerStrategyThreadCount)
+  {
+    this.threadCount = costBalancerStrategyThreadCount;
+  }
 
   @Override
   public BalancerStrategy createBalancerStrategy(DateTime referenceTimestamp)
   {
-    return new CostBalancerMultithreadStrategy(referenceTimestamp);
+    return new CostBalancerMultithreadStrategy(referenceTimestamp, threadCount);
   }
 }

server/src/main/java/io/druid/server/coordinator/DruidCoordinator.java

@@ -745,7 +745,7 @@ public class DruidCoordinator
         else if (RANDOM.equals(config.getCoordinatorBalancerStrategy()))
           factory= new RandomBalancerStrategyFactory();
         else if (COST_MULTI.equals(config.getCoordinatorBalancerStrategy()))
-          factory = new CostBalancerMultithreadStrategyFactory();
+          factory = new CostBalancerMultithreadStrategyFactory(config.getCostBalancerStrategyThreadCount());
 
         // Do coordinator stuff.
         DruidCoordinatorRuntimeParams params =

server/src/main/java/io/druid/server/coordinator/DruidCoordinatorConfig.java

@@ -46,6 +46,10 @@ public abstract class DruidCoordinatorConfig
   @Default("cost")
   public abstract String getCoordinatorBalancerStrategy();
 
+  @Config("druid.coordinator.cost.balancer.threads")
+  @Default("4")
+  public abstract int getCostBalancerStrategyThreadCount();
+
   @Config("druid.coordinator.merge.on")
   public boolean isMergeSegments()
   {

server/src/test/java/io/druid/server/coordination/CostBalancerStrategyTest.java

@@ -3,6 +3,8 @@ package io.druid.server.coordination;
 import com.google.api.client.util.Lists;
 import com.google.common.collect.Maps;
 import io.druid.client.DruidServer;
+import io.druid.server.coordinator.BalancerStrategy;
+import io.druid.server.coordinator.CostBalancerMultithreadStrategy;
 import io.druid.server.coordinator.CostBalancerStrategy;
 import io.druid.server.coordinator.LoadQueuePeonTester;
 import io.druid.server.coordinator.ServerHolder;
@@ -77,7 +79,7 @@ public class CostBalancerStrategyTest
   public void testCostBalancerStrategy() throws InterruptedException {
     DataSegment segment = getSegment(1000);
 
-    CostBalancerStrategy strategy = new CostBalancerStrategy(DateTime.now(DateTimeZone.UTC));
+    BalancerStrategy strategy = new CostBalancerStrategy(DateTime.now(DateTimeZone.UTC));
     ServerHolder holder = strategy.findNewSegmentHomeReplicator(segment, serverHolderList);
     Assert.assertNotNull("Should be able to find a place for new segment!!", holder);
     Assert.assertEquals("Best Server should be BEST_SERVER", "BEST_SERVER", holder.getServer().getName());
@@ -87,7 +89,7 @@ public class CostBalancerStrategyTest
   public void testCostBalancerMultithreadStrategy() throws InterruptedException {
     DataSegment segment = getSegment(1000);
 
-    CostBalancerStrategy strategy = new CostBalancerStrategy(DateTime.now(DateTimeZone.UTC));
+    BalancerStrategy strategy = new CostBalancerMultithreadStrategy(DateTime.now(DateTimeZone.UTC), 8);
     ServerHolder holder = strategy.findNewSegmentHomeReplicator(segment, serverHolderList);
     Assert.assertNotNull("Should be able to find a place for new segment!!", holder);
     Assert.assertEquals("Best Server should be BEST_SERVER", "BEST_SERVER", holder.getServer().getName());

server/src/test/java/io/druid/server/coordinator/DruidCoordinatorTest.java

@@ -101,6 +101,12 @@ public class DruidCoordinatorTest
           {
             return null;
           }
+
+          @Override
+          public int getCostBalancerStrategyThreadCount()
+          {
+            return 1;
+          }
         },
         new ZkPathsConfig(){