azkaban-aplcache

diff --git a/azkaban-common/src/main/java/azkaban/executor/dispatcher/CandidateComparator.java b/azkaban-common/src/main/java/azkaban/executor/dispatcher/CandidateComparator.java
index 1b6db3d..8ff86eb 100644
--- a/azkaban-common/src/main/java/azkaban/executor/dispatcher/CandidateComparator.java
+++ b/azkaban-common/src/main/java/azkaban/executor/dispatcher/CandidateComparator.java
@@ -43,13 +43,24 @@ public abstract class CandidateComparator<T> implements Comparator<T> {
    * */
   public abstract String getName();
 
+  /** differentiate method which will kick in when the comparator list generated an equality result for
+   *  both sides. the differentiate method will try best to make sure a stable result is returned.
+   * */
+  protected boolean differentiate(T object1, T object2){
+    if (object2 == null) return true;
+    if (object1 == null) return false;
+
+    return object1.hashCode() >= object2.hashCode();
+  }
+
   /** function to register a factorComparator to the internal Map for future reference.
    * @param factorComparator : the comparator object to be registered.
    * */
   protected void registerFactorComparator(FactorComparator<T> comparator){
       if (null == comparator ||
           Integer.MAX_VALUE - this.getTotalWeight() < comparator.getWeight() ) {
-        logger.info("skipping registerFactorComparator as the comaractor is null or has an invalid weight value.");
+        logger.info(
+            "skipping registerFactorComparator as the comaractor is null or has an invalid weight value.");
         return;
       }
 
@@ -59,7 +70,6 @@ public abstract class CandidateComparator<T> implements Comparator<T> {
           comparator.getFactorName(), comparator.getWeight()));
   }
 
-
   /** function update the weight of a specific registered factorCompartor.
    * @param factorName : the name of the registered factorComparator to adjust.
    * @param weight:      the new weight value to be adjusted to.
@@ -72,7 +82,8 @@ public abstract class CandidateComparator<T> implements Comparator<T> {
         factorName == "" ||
         weight < 0 ||
         Integer.MAX_VALUE - this.getTotalWeight() < weight){
-      logger.info("skipping adjustFactorWeight as one or more of the input parameters are invalid");
+      logger.info(
+          "skipping adjustFactorWeight as one or more of the input parameters are invalid");
       return -1;
     }
 
@@ -80,7 +91,9 @@ public abstract class CandidateComparator<T> implements Comparator<T> {
 
     // shortcut if the key doesn't exist.
     if (null == value){
-      logger.info(String.format("unable to udpate weight as the specified factorName %s doesn't exist",factorName));
+      logger.info(String.format(
+          "unable to udpate weight as the specified factorName %s doesn't exist",
+          factorName));
       return -1;
     }
 
@@ -153,6 +166,14 @@ public abstract class CandidateComparator<T> implements Comparator<T> {
       logger.info(String.format("[Factor: %s] compare result : %s (current score %s vs %s)",
           comparator.getFactorName(), result, result1, result2));
     }
+
+    // in case of same score, user differentiator to stabilize the result.
+    if (result1 == result2){
+      boolean result = this.differentiate(object1, object2);
+      logger.info("[Differentiator] differentiator chose " + (result?  object1.toString(): object2.toString()));
+      if (result) result1++; else result2++;
+    }
+
     logger.info(String.format("Result : %s vs %s ",result1,result2));
     return new Pair<Integer,Integer>(result1,result2);
   }

diff --git a/azkaban-common/src/test/java/azkaban/executor/DispatcherTest.java b/azkaban-common/src/test/java/azkaban/executor/DispatcherTest.java
index d5d5f47..a5a5cf8 100644
--- a/azkaban-common/src/test/java/azkaban/executor/DispatcherTest.java
+++ b/azkaban-common/src/test/java/azkaban/executor/DispatcherTest.java
@@ -178,6 +178,13 @@ public class DispatcherTest {
       return "MockComparator";
     }
 
+    @Override
+    protected boolean differentiate(MockExecutorObject object1, MockExecutorObject object2){
+      if (object2 == null) return true;
+      if (object1 == null) return false;
+      return object1.name.compareTo(object2.name) >= 0;
+    }
+
     public MockComparator(){
     }
 
@@ -242,8 +249,23 @@ public class DispatcherTest {
     executorList.add(new MockExecutorObject("Executor8",8080,50.0,2048,1,new Date(), 90, 3200));
     executorList.add(new MockExecutorObject("Executor9",8080,50.0,2050,5,new Date(), 90, 4200));
     executorList.add(new MockExecutorObject("Executor10",8080,00.0,1024,1,new Date(), 90, 3200));
-    executorList.add(new MockExecutorObject("Executor11",8080,40.0,3096,3,new Date(), 90, 2400));
-    executorList.add(new MockExecutorObject("Executor12",8080,50.0,2050,5,new Date(), 60, 7200));
+    executorList.add(new MockExecutorObject("Executor11",8080,20.0,2096,3,new Date(), 90, 2400));
+    executorList.add(new MockExecutorObject("Executor12",8080,90.0,2050,5,new Date(), 60, 2500));
+
+
+    // make sure each time the order is different.
+    Collections.shuffle(this.executorList);
+  }
+
+  private MockExecutorObject  getExecutorByName(String name){
+    MockExecutorObject returnVal = null;
+    for (MockExecutorObject item : this.executorList){
+      if (item.name.equals(name)){
+        returnVal = item;
+        break;
+      }
+    }
+    return returnVal;
   }
 
   @After
@@ -260,7 +282,7 @@ public class DispatcherTest {
       mFilter.registerFilterforRemainingMemory();
 
       // expect true.
-      boolean result = mFilter.check(this.executorList.get(0), dispatchingObj);
+      boolean result = mFilter.check(this.getExecutorByName("Executor1"), dispatchingObj);
       /*
        1 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor1' with factor filter for 'Mockfilter'
        1 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
@@ -269,7 +291,7 @@ public class DispatcherTest {
       Assert.assertTrue(result);
 
       //expect true.
-      result = mFilter.check(this.executorList.get(2), dispatchingObj);
+      result = mFilter.check(this.getExecutorByName("Executor3"), dispatchingObj);
       /*
       1 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor3' with factor filter for 'Mockfilter'
       2 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
@@ -279,7 +301,7 @@ public class DispatcherTest {
 
       // add the priority filter.
       mFilter.registerFilterforPriority();
-      result = mFilter.check(this.executorList.get(2), dispatchingObj);
+      result = mFilter.check(this.getExecutorByName("Executor3"), dispatchingObj);
       // expect false, for priority.
       /*
       2 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor3' with factor filter for 'Mockfilter'
@@ -293,7 +315,7 @@ public class DispatcherTest {
       mFilter.registerFilterforRemainingTmpSpace();
 
       // expect pass.
-      result = mFilter.check(this.executorList.get(1), dispatchingObj);
+      result = mFilter.check(this.getExecutorByName("Executor2"), dispatchingObj);
       /*
       3 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor2' with factor filter for 'Mockfilter'
       3 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
@@ -304,7 +326,7 @@ public class DispatcherTest {
       Assert.assertTrue(result);
 
       // expect false, remaining tmp, priority will also fail but the logic shortcuts when the Tmp size check Fails.
-      result = mFilter.check(this.executorList.get(7), dispatchingObj);
+      result = mFilter.check(this.getExecutorByName("Executor8"), dispatchingObj);
       /*
       4 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor8' with factor filter for 'Mockfilter'
       4 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
@@ -323,59 +345,94 @@ public class DispatcherTest {
     MockExecutorObject nextExecutor = Collections.max(this.executorList, comparator);
 
     // expect the first item to be selected, memory wise it is the max.
-    Assert.assertEquals(this.executorList.get(10),nextExecutor);
+    Assert.assertEquals(this.getExecutorByName("Executor11"),nextExecutor);
 
     // add the priority factor.
     // expect again the #9 item to be selected.
     comparator.registerComparerForPriority(6);
     nextExecutor = Collections.max(this.executorList, comparator);
     /*
-      10 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor2' with 'Executor1',  total weight = 11
-      10 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 0 (current score 0 vs 0)
-      11 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 0 vs 6)
-      11 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 6
-      11 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor3' with 'Executor1',  total weight = 11
+      2 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor8' with 'Executor2',  total weight = 5 
+      2 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 0 (current score 0 vs 0)
+      2 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Differentiator] differentiator chose Executor8
+      3 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 1 vs 0 
+      3 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor3' with 'Executor8',  total weight = 5 
+      4 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
+      4 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5 
+      4 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor9' with 'Executor8',  total weight = 5 
+      5 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 1 (current score 5 vs 0)
+      5 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 5 vs 0 
+      5 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor11' with 'Executor9',  total weight = 5 
+      6 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 1 (current score 5 vs 0)
+      6 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 5 vs 0 
+      6 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor12' with 'Executor11',  total weight = 5 
+      6 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
+      7 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5 
+      7 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor10' with 'Executor11',  total weight = 5 
+      7 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
+      8 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5 
+      8 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor1' with 'Executor11',  total weight = 5 
+      8 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
+      9 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5 
+      9 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor7' with 'Executor11',  total weight = 5 
+      9 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
+      9 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5 
+      10 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor6' with 'Executor11',  total weight = 5 
+      10 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
+      10 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5 
+      11 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor5' with 'Executor11',  total weight = 5 
       11 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
-      11 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 0 vs 11)
-      11 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 11
-      12 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor4' with 'Executor1',  total weight = 11
-      12 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 0 (current score 0 vs 0)
-      12 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 0 vs 6)
-      12 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 6
-      13 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor5' with 'Executor1',  total weight = 11
-      13 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
-      13 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 0 (current score 0 vs 5)
-      14 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5
-      14 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor6' with 'Executor1',  total weight = 11
-      14 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
-      14 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 0 (current score 0 vs 5)
-      14 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5
-      14 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor7' with 'Executor1',  total weight = 11
+      11 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5 
+      12 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor4' with 'Executor11',  total weight = 5 
+      12 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
+      12 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5 
+      14 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Factor comparator added for 'Priority'. Weight = '6'
+      14 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor8' with 'Executor2',  total weight = 11 
+      14 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 0 (current score 0 vs 0)
+      14 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 0 vs 6)
+      14 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 6 
+      15 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor3' with 'Executor2',  total weight = 11 
       15 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
-      15 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 0 (current score 0 vs 5)
-      15 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5
-      15 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor8' with 'Executor1',  total weight = 11
-      15 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 0 (current score 0 vs 0)
-      16 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 0 vs 6)
-      16 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 6
-      16 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor9' with 'Executor1',  total weight = 11
+      15 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 0 vs 11)
+      15 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 11 
+      15 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor9' with 'Executor2',  total weight = 11 
+      15 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 1 (current score 5 vs 0)
+      16 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 1 (current score 11 vs 0)
+      16 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 11 vs 0 
+      16 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor11' with 'Executor9',  total weight = 11 
       16 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 1 (current score 5 vs 0)
-      16 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 0 (current score 5 vs 0)
-      16 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 5 vs 0
-      17 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor10' with 'Executor9',  total weight = 11
+      16 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 5 vs 6)
+      16 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 5 vs 6 
+      16 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor12' with 'Executor9',  total weight = 11 
+      17 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 1 (current score 5 vs 0)
+      17 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 0 (current score 5 vs 0)
+      17 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 5 vs 0 
+      17 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor10' with 'Executor12',  total weight = 11 
       17 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
       17 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 0 vs 11)
-      17 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 11
-      18 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor11' with 'Executor9',  total weight = 11
-      18 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 1 (current score 5 vs 0)
-      18 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 5 vs 6)
-      19 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 5 vs 6
-      19 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor12' with 'Executor9',  total weight = 11
-      19 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 0 (current score 0 vs 0)
-      19 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 0 (current score 0 vs 0)
-      20 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 0
+      18 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 11 
+      18 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor1' with 'Executor12',  total weight = 11 
+      18 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
+      18 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 0 (current score 0 vs 5)
+      18 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5 
+      18 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor7' with 'Executor12',  total weight = 11 
+      19 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
+      19 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 0 (current score 0 vs 5)
+      19 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5 
+      19 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor6' with 'Executor12',  total weight = 11 
+      19 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
+      19 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 0 (current score 0 vs 5)
+      19 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5 
+      20 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor5' with 'Executor12',  total weight = 11 
+      20 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
+      20 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 0 (current score 0 vs 5)
+      20 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5 
+      20 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor4' with 'Executor12',  total weight = 11 
+      20 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : -1 (current score 0 vs 5)
+      21 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 0 vs 11)
+      21 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 11 
      * */
-    Assert.assertEquals(this.executorList.get(8),nextExecutor);
+    Assert.assertEquals(this.getExecutorByName("Executor12"),nextExecutor);
 
     // add the remaining space factor.
     // expect the #12 item to be returned.
@@ -438,7 +495,7 @@ public class DispatcherTest {
       34 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: RemainingTmp] compare result : 1 (current score 3 vs 0)
       34 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 3 vs 0
      * */
-    Assert.assertEquals(this.executorList.get(11),nextExecutor);
+    Assert.assertEquals(this.getExecutorByName("Executor12"),nextExecutor);
   }
 
   @Test
@@ -464,84 +521,77 @@ public class DispatcherTest {
     // expected selection = #12
     MockExecutorObject nextExecutor = morkDispatcher.getNext(this.executorList, dispatchingObj);
     /*
-     *  9 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start candidate selection logic.
-        9 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - candidate count before filtering: 12
-        10 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor1' with factor filter for 'Mockfilter'
-        10 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
-        10 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
-        10 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : true
-        11 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredProprity] filter result : true
-        11 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : true
-        11 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor2' with factor filter for 'Mockfilter'
-        11 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
-        11 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
-        12 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : true
-        12 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredProprity] filter result : true
-        12 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : true
-        12 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor3' with factor filter for 'Mockfilter'
-        12 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
-        13 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
-        13 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : true
-        13 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredProprity] filter result : false
-        13 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
-        13 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor4' with factor filter for 'Mockfilter'
-        13 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
-        14 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
-        14 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : true
-        14 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredProprity] filter result : true
-        14 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : true
-        14 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor5' with factor filter for 'Mockfilter'
-        15 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : false
-        15 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
-        15 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor6' with factor filter for 'Mockfilter'
-        15 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : false
-        15 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
-        15 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor7' with factor filter for 'Mockfilter'
-        16 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : false
-        16 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
-        16 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor8' with factor filter for 'Mockfilter'
-        16 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
-        16 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
-        17 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : false
-        17 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
-        17 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor9' with factor filter for 'Mockfilter'
-        17 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
-        17 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
-        17 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : false
-        18 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
-        18 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor10' with factor filter for 'Mockfilter'
-        18 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : false
-        18 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
-        18 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor11' with factor filter for 'Mockfilter'
-        18 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
-        18 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
-        18 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : false
-        19 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
-        19 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor12' with factor filter for 'Mockfilter'
-        19 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
-        19 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
-        19 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : true
-        19 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredProprity] filter result : true
-        19 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : true
-        20 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - candidate count after filtering: 4
-        20 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor2' with 'Executor1',  total weight = 11
-        20 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 0 (current score 0 vs 0)
-        20 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 0 vs 5)
-        21 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: RemainingTmp] compare result : 0 (current score 0 vs 5)
-        21 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5
-        21 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor4' with 'Executor1',  total weight = 11
-        22 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 0 (current score 0 vs 0)
-        22 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 0 vs 5)
-        22 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: RemainingTmp] compare result : 0 (current score 0 vs 5)
-        22 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5
-        22 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor12' with 'Executor1',  total weight = 11
-        23 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 1 (current score 3 vs 0)
-        23 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 0 (current score 3 vs 0)
-        23 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: RemainingTmp] compare result : 1 (current score 6 vs 0)
-        23 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 6 vs 0
-        23 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - candidate selected Executor12
+     *  4743 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start candidate selection logic.
+        4744 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - candidate count before filtering: 12
+        4745 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor2' with factor filter for 'Mockfilter'
+        4746 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
+        4747 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
+        4747 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : true
+        4748 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredProprity] filter result : true
+        4749 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : true
+        4749 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor12' with factor filter for 'Mockfilter'
+        4750 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
+        4751 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : false
+        4751 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
+        4752 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor11' with factor filter for 'Mockfilter'
+        4752 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
+        4753 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
+        4753 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : false
+        4753 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
+        4754 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor7' with factor filter for 'Mockfilter'
+        4754 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : false
+        4754 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
+        4755 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor5' with factor filter for 'Mockfilter'
+        4755 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : false
+        4755 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
+        4756 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor9' with factor filter for 'Mockfilter'
+        4756 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
+        4756 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
+        4757 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : false
+        4757 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
+        4757 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor8' with factor filter for 'Mockfilter'
+        4757 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
+        4758 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
+        4758 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : false
+        4758 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
+        4759 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor1' with factor filter for 'Mockfilter'
+        4759 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
+        4759 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
+        4760 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : true
+        4760 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredProprity] filter result : true
+        4760 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : true
+        4761 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor4' with factor filter for 'Mockfilter'
+        4761 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
+        4761 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
+        4762 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : true
+        4762 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredProprity] filter result : true
+        4762 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : true
+        4763 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor6' with factor filter for 'Mockfilter'
+        4763 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : false
+        4763 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
+        4764 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor3' with factor filter for 'Mockfilter'
+        4764 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : true
+        4764 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredTotalMemory] filter result : true
+        4765 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingTmpSpace] filter result : true
+        4765 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredProprity] filter result : false
+        4765 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
+        4765 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - start checking 'Executor10' with factor filter for 'Mockfilter'
+        4766 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - [Factor: requiredRemainingMemory] filter result : false
+        4766 [main] INFO azkaban.executor.dispatcher.CandidateFilter  - Final checking result : false
+        4766 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - candidate count after filtering: 3
+        4767 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor1' with 'Executor2',  total weight = 11
+        4767 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 0 (current score 0 vs 0)
+        4768 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : 1 (current score 5 vs 0)
+        4768 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: RemainingTmp] compare result : 0 (current score 5 vs 0)
+        4769 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 5 vs 0
+        4770 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - start comparing 'Executor4' with 'Executor1',  total weight = 11
+        4771 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Memory] compare result : 0 (current score 0 vs 0)
+        4771 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: Priority] compare result : -1 (current score 0 vs 5)
+        4772 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - [Factor: RemainingTmp] compare result : 0 (current score 0 vs 5)
+        4772 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - Result : 0 vs 5
+        4772 [main] INFO azkaban.executor.dispatcher.CandidateComparator  - candidate selected Executor1
      */
-    Assert.assertEquals( this.executorList.get(11),nextExecutor);
+    Assert.assertEquals(this.getExecutorByName("Executor1"),nextExecutor);
 
    // remaining memory 11500, total memory 3095, remainingTmpSpace 14200, priority 2.
    dispatchingObj = new MockFlowObject("flow1",3096, 1500,14200,2);
@@ -569,17 +619,17 @@ public class DispatcherTest {
 
     MockFlowObject  dispatchingObj = new MockFlowObject("flow1",100, 100,100,3);
     MockExecutorObject executor = morkDispatcher.getNext(this.executorList, dispatchingObj);
-    Assert.assertEquals(this.executorList.get(11), executor);
+    Assert.assertEquals(this.getExecutorByName("Executor9"), executor);
 
     // adjusted the weight for memory to 10, therefore item #11 should be returned.
     morkDispatcher.getComparator().adjustFactorWeight("Memory", 10);
     executor = morkDispatcher.getNext(this.executorList, dispatchingObj);
-    Assert.assertEquals(this.executorList.get(10), executor);
-    
+    Assert.assertEquals(this.getExecutorByName("Executor11"), executor);
+
     // adjusted the weight for memory back to 1, therefore item #12 should be returned.
     morkDispatcher.getComparator().adjustFactorWeight("Memory", 1);
     executor = morkDispatcher.getNext(this.executorList, dispatchingObj);
-    Assert.assertEquals(this.executorList.get(11), executor);
+    Assert.assertEquals(this.getExecutorByName("Executor9"), executor);
 
   }
 }