Add a new equal distribution strategy for assigning tasks

docs/content/Indexing-Service-Config.md

@@ -139,7 +139,7 @@ Issuing a GET request at the same URL will return the current worker config spec
 
 |Property|Description|Default|
 |--------|-----------|-------|
-|`selectStrategy`|How to assign tasks to middlemanagers. Choices are `fillCapacity` and `fillCapacityWithAffinity`.|fillCapacity|
+|`selectStrategy`|How to assign tasks to middlemanagers. Choices are `fillCapacity`, `fillCapacityWithAffinity`, and `equalDistribution`.|fillCapacity|
 |`autoScaler`|Only used if autoscaling is enabled. See below.|null|
 
 #### Worker Select Strategy
@@ -163,6 +163,15 @@ An affinity config can be provided.
 
 Tasks will try to be assigned to preferred workers. Fill capacity strategy is used if no preference for a datasource specified.
 
+##### Equal Distribution
+
+The workers with the least amount of tasks is assigned the task.
+
+|Property|Description|Default|
+|--------|-----------|-------|
+|`type`|`equalDistribution`.|fillCapacity|
+
+
 #### Autoscaler
 
 Amazon's EC2 is currently the only supported autoscaler.

indexing-service/src/main/java/io/druid/indexing/overlord/setup/EqualDistributionWorkerSelectStrategy.java

@@ -0,0 +1,65 @@
+/*
+ * Druid - a distributed column store.
+ * Copyright (C) 2012, 2013  Metamarkets Group Inc.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
+
+package io.druid.indexing.overlord.setup;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Sets;
+import com.google.common.primitives.Ints;
+import io.druid.indexing.common.task.Task;
+import io.druid.indexing.overlord.ImmutableZkWorker;
+import io.druid.indexing.overlord.config.RemoteTaskRunnerConfig;
+
+import java.util.Comparator;
+import java.util.TreeSet;
+
+/**
+ */
+public class EqualDistributionWorkerSelectStrategy implements WorkerSelectStrategy
+{
+  @Override
+  public Optional<ImmutableZkWorker> findWorkerForTask(
+      RemoteTaskRunnerConfig config, ImmutableMap<String, ImmutableZkWorker> zkWorkers, Task task
+  )
+  {
+    final TreeSet<ImmutableZkWorker> sortedWorkers = Sets.newTreeSet(
+        new Comparator<ImmutableZkWorker>()
+        {
+          @Override
+          public int compare(
+              ImmutableZkWorker zkWorker, ImmutableZkWorker zkWorker2
+          )
+          {
+            return -Ints.compare(zkWorker2.getCurrCapacityUsed(), zkWorker.getCurrCapacityUsed());
+          }
+        }
+    );
+    sortedWorkers.addAll(zkWorkers.values());
+    final String minWorkerVer = config.getMinWorkerVersion();
+
+    for (ImmutableZkWorker zkWorker : sortedWorkers) {
+      if (zkWorker.canRunTask(task) && zkWorker.isValidVersion(minWorkerVer)) {
+        return Optional.of(zkWorker);
+      }
+    }
+
+    return Optional.absent();
+  }
+}

indexing-service/src/main/java/io/druid/indexing/overlord/setup/WorkerSelectStrategy.java

@@ -33,7 +33,8 @@ import io.druid.indexing.overlord.config.RemoteTaskRunnerConfig;
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", defaultImpl = FillCapacityWorkerSelectStrategy.class)
 @JsonSubTypes(value = {
     @JsonSubTypes.Type(name = "fillCapacity", value = FillCapacityWorkerSelectStrategy.class),
-    @JsonSubTypes.Type(name = "fillCapacityWithAffinity", value = FillCapacityWithAffinityWorkerSelectStrategy.class)
+    @JsonSubTypes.Type(name = "fillCapacityWithAffinity", value = FillCapacityWithAffinityWorkerSelectStrategy.class),
+    @JsonSubTypes.Type(name = "equalDistribution", value = EqualDistributionWorkerSelectStrategy.class)
 })
 public interface WorkerSelectStrategy
 {

indexing-service/src/test/java/io/druid/indexing/overlord/setup/EqualDistributionWorkerSelectStrategyTest.java

@@ -0,0 +1,70 @@
+/*
+ * Druid - a distributed column store.
+ * Copyright (C) 2012, 2013  Metamarkets Group Inc.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
+
+package io.druid.indexing.overlord.setup;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Sets;
+import io.druid.indexing.common.task.NoopTask;
+import io.druid.indexing.overlord.ImmutableZkWorker;
+import io.druid.indexing.overlord.config.RemoteTaskRunnerConfig;
+import io.druid.indexing.worker.Worker;
+import junit.framework.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+
+import static org.junit.Assert.*;
+
+public class EqualDistributionWorkerSelectStrategyTest
+{
+
+  @Test
+  public void testFindWorkerForTask() throws Exception
+  {
+    final EqualDistributionWorkerSelectStrategy strategy = new EqualDistributionWorkerSelectStrategy();
+
+    Optional<ImmutableZkWorker> optional = strategy.findWorkerForTask(
+        new RemoteTaskRunnerConfig(),
+        ImmutableMap.of(
+            "lhost",
+            new ImmutableZkWorker(
+                new Worker("lhost", "lhost", 1, "v1"), 0,
+                Sets.<String>newHashSet()
+            ),
+            "localhost",
+            new ImmutableZkWorker(
+                new Worker("localhost", "localhost", 1, "v1"), 1,
+                Sets.<String>newHashSet()
+            )
+        ),
+        new NoopTask(null, 1, 0, null, null)
+        {
+          @Override
+          public String getDataSource()
+          {
+            return "foo";
+          }
+        }
+    );
+    ImmutableZkWorker worker = optional.get();
+    Assert.assertEquals("lhost", worker.getWorker().getHost());
+  }
+}