azkaban-aplcache

diff --git a/azkaban-common/src/main/java/azkaban/executor/selector/CandidateComparator.java b/azkaban-common/src/main/java/azkaban/executor/selector/CandidateComparator.java
index 8ef2c76..2f89129 100644
--- a/azkaban-common/src/main/java/azkaban/executor/selector/CandidateComparator.java
+++ b/azkaban-common/src/main/java/azkaban/executor/selector/CandidateComparator.java
@@ -25,10 +25,12 @@ import org.apache.log4j.Logger;
 
 import azkaban.utils.Pair;
 
-/** Abstract class for a candidate comparator.
+/**
+ * <pre>
+ *  Abstract class for a candidate comparator.
  *  this class contains implementation of most of the core logics. Implementing classes is expected only to
  *  register factor comparators using the provided register function.
- *
+ * <pre>
  */
 public abstract class CandidateComparator<T> implements Comparator<T> {
   protected static Logger logger = Logger.getLogger(CandidateComparator.class);
@@ -85,7 +87,9 @@ public abstract class CandidateComparator<T> implements Comparator<T> {
     return totalWeight;
   }
 
-  /** function to actually calculate the scores for the two objects that are being compared.
+  /**
+   * <pre>
+   * function to actually calculate the scores for the two objects that are being compared.
    *  the comparison follows the following logic -
    *  1. if both objects are equal return 0 score for both.
    *  2. if one side is null, the other side gets all the score.
@@ -94,6 +98,10 @@ public abstract class CandidateComparator<T> implements Comparator<T> {
    *     added to the wining side, if equal, no value will be added to either side.
    *  4. final result will be returned in a Pair container.
    *
+   * </pre>
+   * @param object1  the first  object (left side)  to be compared.
+   * @param object2  the second object (right side) to be compared.
+   * @return a pair structure contains the score for both sides.
    * */
   public Pair<Integer,Integer> getComparisonScore(T object1, T object2){
     logger.info(String.format("start comparing '%s' with '%s',  total weight = %s ",

diff --git a/azkaban-common/src/main/java/azkaban/executor/selector/CandidateSelector.java b/azkaban-common/src/main/java/azkaban/executor/selector/CandidateSelector.java
index 53cb1bb..5cae9ef 100644
--- a/azkaban-common/src/main/java/azkaban/executor/selector/CandidateSelector.java
+++ b/azkaban-common/src/main/java/azkaban/executor/selector/CandidateSelector.java
@@ -17,9 +17,8 @@
 package azkaban.executor.selector;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
-import java.util.List;
-
 import org.apache.log4j.Logger;
 
 /** Implementation of the CandidateSelector.
@@ -43,7 +42,7 @@ public class CandidateSelector<K extends Comparable<K>, V> implements Selector<K
   }
 
   @Override
-  public K getBest(List<K> candidateList, V dispatchingObject) {
+  public K getBest(Collection<K> candidateList, V dispatchingObject) {
 
      // shortcut if the candidateList is empty.
      if ( null == candidateList || candidateList.size() == 0){
@@ -55,7 +54,7 @@ public class CandidateSelector<K extends Comparable<K>, V> implements Selector<K
      logger.info(String.format("candidate count before filtering: %s", candidateList.size()));
 
      // to keep the input untouched, we will form up a new list based off the filtering result.
-     List<K> filteredList = new ArrayList<K>();
+     Collection<K> filteredList = new ArrayList<K>();
 
      if (null != this.filter){
        for (K candidateInfo : candidateList){

diff --git a/azkaban-common/src/main/java/azkaban/executor/selector/ExecutorFilter.java b/azkaban-common/src/main/java/azkaban/executor/selector/ExecutorFilter.java
index 031650e..19baa10 100644
--- a/azkaban-common/src/main/java/azkaban/executor/selector/ExecutorFilter.java
+++ b/azkaban-common/src/main/java/azkaban/executor/selector/ExecutorFilter.java
@@ -16,8 +16,8 @@
 
 package azkaban.executor.selector;
 
+import java.util.Collection;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -45,10 +45,11 @@ public final class ExecutorFilter extends CandidateFilter<Executor, ExecutableFl
   private static final String MINIMUMFREEMEMORY_FILTER_NAME = "MinimunFreeMemory";
   private static final String CPUSTATUS_FILTER_NAME = "CpuStatus";
 
-  /**
+  /**<pre>
    * static initializer of the class.
    * We will build the filter repository here.
    * when a new filter is added, please do remember to register it here.
+   * </pre>
    * */
   static {
     filterRepository = new HashMap<String, FactorFilter<Executor, ExecutableFlow>>();
@@ -61,7 +62,7 @@ public final class ExecutorFilter extends CandidateFilter<Executor, ExecutableFl
    * constructor of the ExecutorFilter.
    * @param filterList   the list of filter to be registered, the parameter must be a not-empty and valid list object.
    * */
-  public ExecutorFilter(List<String> filterList) {
+  public ExecutorFilter(Collection<String> filterList) {
     // shortcut if the filter list is invalid. A little bit ugly to have to throw in constructor.
     if (null == filterList || filterList.size() == 0){
       logger.error("failed to initialize executor filter as the passed filter list is invalid or empty.");
@@ -113,11 +114,12 @@ public final class ExecutorFilter extends CandidateFilter<Executor, ExecutableFl
     });
   }
 
-  /**
+  /**<pre>
    * function to register the static Minimum Reserved Memory filter.
    * NOTE : this is a static filter which means the filter will be filtering based on the system standard which is not
    *        Coming for the passed flow.
    *        This filter will filter out any executors that has the remaining  memory below 6G
+   *</pre>
    * */
   private static FactorFilter<Executor, ExecutableFlow> getMinimumReservedMemoryFilter(){
     return FactorFilter.create(MINIMUMFREEMEMORY_FILTER_NAME, new FactorFilter.Filter<Executor, ExecutableFlow>() {
@@ -142,11 +144,12 @@ public final class ExecutorFilter extends CandidateFilter<Executor, ExecutableFl
 
 
   /**
+   * <pre>
    * function to register the static Minimum Reserved Memory filter.
-   * NOTE : <pre> this is a static filter which means the filter will be filtering based on the system standard which
+   * NOTE :  this is a static filter which means the filter will be filtering based on the system standard which
    *        is not Coming for the passed flow.
    *        This filter will filter out any executors that the current CPU usage exceed 95%
-   *        </pre>
+   * </pre>
    * */
   private static FactorFilter<Executor, ExecutableFlow> getCpuStatusFilter(){
     return FactorFilter.create(CPUSTATUS_FILTER_NAME, new FactorFilter.Filter<Executor, ExecutableFlow>() {

diff --git a/azkaban-common/src/main/java/azkaban/executor/selector/ExecutorSelector.java b/azkaban-common/src/main/java/azkaban/executor/selector/ExecutorSelector.java
index d7236e8..2405c28 100644
--- a/azkaban-common/src/main/java/azkaban/executor/selector/ExecutorSelector.java
+++ b/azkaban-common/src/main/java/azkaban/executor/selector/ExecutorSelector.java
@@ -16,7 +16,7 @@
 
 package azkaban.executor.selector;
 
-import java.util.List;
+import java.util.Collection;
 import java.util.Map;
 
 import azkaban.executor.ExecutableFlow;
@@ -38,7 +38,7 @@ public class ExecutorSelector extends CandidateSelector<Executor, ExecutableFlow
    * @param comparatorList  name/weight pair list of the comparators to be registered ,
    *                        again comparator feature is disabled if a null value is passed.
    * */
-  public ExecutorSelector(List<String> filterList, Map<String,Integer> comparatorList) {
+  public ExecutorSelector(Collection<String> filterList, Map<String,Integer> comparatorList) {
     super(null == filterList || filterList.isEmpty() ?         null : new ExecutorFilter(filterList),
           null == comparatorList || comparatorList.isEmpty() ? null : new ExecutorComparator(comparatorList));
   }

diff --git a/azkaban-common/src/main/java/azkaban/executor/selector/Selector.java b/azkaban-common/src/main/java/azkaban/executor/selector/Selector.java
index 605fd28..a56b41a 100644
--- a/azkaban-common/src/main/java/azkaban/executor/selector/Selector.java
+++ b/azkaban-common/src/main/java/azkaban/executor/selector/Selector.java
@@ -16,7 +16,7 @@
 
 package azkaban.executor.selector;
 
-import java.util.List;
+import java.util.Collection;
 
 
 /**<pre>
@@ -34,7 +34,7 @@ public interface Selector <K extends Comparable<K>,V> {
    *  @param  dispatchingObject : the object to be dispatched .
    *  @return candidate from the candidate list that suits best for the dispatching object.
    * */
-  public K getBest(List<K> candidateList, V dispatchingObject);
+  public K getBest(Collection<K> candidateList, V dispatchingObject);
 
   /** Function returns the name of the current Dispatcher
    *  @return name of the dispatcher.