fix bug with provision

merger/src/main/java/com/metamx/druid/merger/coordinator/RemoteTaskRunner.java

@@ -439,7 +439,7 @@ public class RemoteTaskRunner implements TaskRunner
 
       if (workerQueue.isEmpty()) {
         log.makeAlert("There are no worker nodes with capacity to run task!").emit();
-        strategy.provisionIfNeeded(zkWorkers);
+        strategy.provision(zkWorkers);
         return null;
       }
 

merger/src/main/java/com/metamx/druid/merger/coordinator/scaling/NoopScalingStrategy.java

@@ -9,6 +9,7 @@ import com.metamx.common.guava.FunctionalIterable;
 import com.metamx.druid.merger.coordinator.WorkerWrapper;
 import com.metamx.druid.merger.coordinator.config.S3AutoScalingStrategyConfig;
 import com.metamx.emitter.EmittingLogger;
+import org.joda.time.DateTime;
 
 import java.util.Comparator;
 import java.util.Map;
@@ -35,7 +36,7 @@ public class NoopScalingStrategy implements ScalingStrategy
   }
 
   @Override
-  public void provisionIfNeeded(Map<String, WorkerWrapper> zkWorkers)
+  public void provision(Map<String, WorkerWrapper> zkWorkers)
   {
     synchronized (lock) {
       if (currentlyProvisioning != null) {
@@ -48,18 +49,6 @@ public class NoopScalingStrategy implements ScalingStrategy
         }
       }
 
-      Iterable<WorkerWrapper> availableWorkers = FunctionalIterable.create(zkWorkers.values()).filter(
-          new Predicate<WorkerWrapper>()
-          {
-            @Override
-            public boolean apply(WorkerWrapper input)
-            {
-              return !input.isAtCapacity();
-            }
-          }
-      );
-
-      if (Iterables.size(availableWorkers) == 0) {
       try {
         log.info("If I were a real strategy I'd create something now");
         currentlyProvisioning = "willNeverBeTrue";
@@ -70,7 +59,6 @@ public class NoopScalingStrategy implements ScalingStrategy
       }
     }
   }
-  }
 
   @Override
   public Instance terminateIfNeeded(Map<String, WorkerWrapper> zkWorkers)
@@ -89,9 +77,9 @@ public class NoopScalingStrategy implements ScalingStrategy
             @Override
             public int compare(WorkerWrapper w1, WorkerWrapper w2)
             {
-              return Ordering.natural()
+              DateTime w1Time = (w1 == null) ? new DateTime(0) : w1.getLastCompletedTaskTime();
-                             .nullsFirst()
+              DateTime w2Time = (w2 == null) ? new DateTime(0) : w2.getLastCompletedTaskTime();
-                             .compare(w1.getLastCompletedTaskTime(), w2.getLastCompletedTaskTime());
+              return w1Time.compareTo(w2Time);
             }
           }
       ).create(

merger/src/main/java/com/metamx/druid/merger/coordinator/scaling/S3AutoScalingStrategy.java

@@ -36,6 +36,7 @@ import com.metamx.common.guava.FunctionalIterable;
 import com.metamx.druid.merger.coordinator.WorkerWrapper;
 import com.metamx.druid.merger.coordinator.config.S3AutoScalingStrategyConfig;
 import com.metamx.emitter.EmittingLogger;
+import org.joda.time.DateTime;
 
 import java.util.Arrays;
 import java.util.Comparator;
@@ -65,7 +66,7 @@ public class S3AutoScalingStrategy implements ScalingStrategy
   }
 
   @Override
-  public void provisionIfNeeded(Map<String, WorkerWrapper> zkWorkers)
+  public void provision(Map<String, WorkerWrapper> zkWorkers)
   {
     synchronized (lock) {
       if (zkWorkers.containsKey(currentlyProvisioning)) {
@@ -80,18 +81,6 @@ public class S3AutoScalingStrategy implements ScalingStrategy
         return;
       }
 
-      Iterable<WorkerWrapper> availableWorkers = FunctionalIterable.create(zkWorkers.values()).filter(
-          new Predicate<WorkerWrapper>()
-          {
-            @Override
-            public boolean apply(WorkerWrapper input)
-            {
-              return !input.isAtCapacity();
-            }
-          }
-      );
-
-      if (Iterables.size(availableWorkers) == 0) {
       try {
         log.info("Creating a new instance");
         RunInstancesResult result = amazonEC2Client.runInstances(
@@ -115,7 +104,6 @@ public class S3AutoScalingStrategy implements ScalingStrategy
       }
     }
   }
-  }
 
   @Override
   public Instance terminateIfNeeded(Map<String, WorkerWrapper> zkWorkers)
@@ -136,7 +124,9 @@ public class S3AutoScalingStrategy implements ScalingStrategy
             @Override
             public int compare(WorkerWrapper w1, WorkerWrapper w2)
             {
-              return w1.getLastCompletedTaskTime().compareTo(w2.getLastCompletedTaskTime());
+              DateTime w1Time = (w1 == null) ? new DateTime(0) : w1.getLastCompletedTaskTime();
+              DateTime w2Time = (w2 == null) ? new DateTime(0) : w2.getLastCompletedTaskTime();
+              return w1Time.compareTo(w2Time);
             }
           }
       ).create(

merger/src/main/java/com/metamx/druid/merger/coordinator/scaling/ScalingStrategy.java

@@ -28,7 +28,7 @@ import java.util.Map;
  */
 public interface ScalingStrategy
 {
-  public void provisionIfNeeded(Map<String, WorkerWrapper> zkWorkers);
+  public void provision(Map<String, WorkerWrapper> zkWorkers);
 
   public Instance terminateIfNeeded(Map<String, WorkerWrapper> zkWorkers);
 }

merger/src/test/java/com/metamx/druid/merger/coordinator/scaling/S3AutoScalingStrategyTest.java

@@ -152,7 +152,7 @@ public class S3AutoScalingStrategyTest
 
     Assert.assertTrue(worker.isAtCapacity());
 
-    strategy.provisionIfNeeded(zkWorkers);
+    strategy.provision(zkWorkers);
 
     worker.getRunningTasks().remove("task1");
     worker.getRunningTasks().remove("task2");