YARN-6610. [YARN-3926] DominantResourceCalculator#getResourceAsValue dominant param is updated to handle multiple resources. Contributed by Daniel Templeton.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java

@@ -17,12 +17,18 @@
 */
 package org.apache.hadoop.yarn.util.resource;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceInformation;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
 import org.apache.hadoop.yarn.util.UnitsConversionUtil;
 
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.Arrays;
 
 /**
  * A {@link ResourceCalculator} which uses the concept of
@@ -48,6 +54,7 @@ import org.apache.hadoop.yarn.util.UnitsConversionUtil;
 @Private
 @Unstable
 public class DominantResourceCalculator extends ResourceCalculator {
+  static final Log LOG = LogFactory.getLog(DominantResourceCalculator.class);
 
   public DominantResourceCalculator() {
   }
@@ -92,7 +99,6 @@ public class DominantResourceCalculator extends ResourceCalculator {
   @Override
   public int compare(Resource clusterResource, Resource lhs, Resource rhs,
       boolean singleType) {
-    
     if (lhs.equals(rhs)) {
       return 0;
     }
@@ -101,55 +107,232 @@ public class DominantResourceCalculator extends ResourceCalculator {
       return this.compare(lhs, rhs);
     }
 
-    float l = getResourceAsValue(clusterResource, lhs, true);
-    float r = getResourceAsValue(clusterResource, rhs, true);
+    // We have to calculate the shares for all resource types for both
+    // resources and then look for which resource has the biggest
+    // share overall.
+    ResourceInformation[] clusterRes = clusterResource.getResources();
+    // If array creation shows up as a time sink, these arrays could be cached
+    // because they're always the same length.
+    double[] lhsShares = new double[clusterRes.length];
+    double[] rhsShares = new double[clusterRes.length];
+    double diff;
+
+    try {
+      if (singleType) {
+        double[] max = new double[2];
+
+        calculateShares(clusterRes, lhs, rhs, lhsShares, rhsShares, max);
+
+        diff = max[0] - max[1];
+      } else if (clusterRes.length == 2) {
+        // Special case to handle the common scenario of only CPU and memory
+        // so the we can optimize for performance
+        diff = calculateSharesForMandatoryResources(clusterRes, lhs, rhs,
+            lhsShares, rhsShares);
+      } else {
+        calculateShares(clusterRes, lhs, rhs, lhsShares, rhsShares);
+
+        Arrays.sort(lhsShares);
+        Arrays.sort(rhsShares);
+
+        diff = compareShares(lhsShares, rhsShares);
+      }
+    } catch (ArrayIndexOutOfBoundsException ex) {
+      StringWriter out = new StringWriter(); // No need to close a StringWriter
+      ex.printStackTrace(new PrintWriter(out));
+
+      LOG.error("A problem was encountered while calculating resource "
+          + "availability that should not occur under normal circumstances. "
+          + "Please report this error to the Hadoop community by opening a "
+          + "JIRA ticket at http://issues.apache.org/jira and including the "
+          + "following information:\n* Exception encountered: " + out + "* "
+          + "Cluster resources: " + Arrays.toString(clusterRes) + "\n* "
+          + "LHS resource: " + Arrays.toString(lhs.getResources()) + "\n* "
+          + "RHS resource: " + Arrays.toString(rhs.getResources()));
+      LOG.error("The resource manager is in an inconsistent state. It is safe "
+          + "for the resource manager to be restarted as the error encountered "
+          + "should be transitive. If high availability is enabled, failing "
+          + "over to a standby resource manager is also safe.");
+      throw new YarnRuntimeException("A problem was encountered while "
+          + "calculating resource availability that should not occur under "
+          + "normal circumstances. Please see the log for more information.",
+          ex);
+    }
+
+    return (int) Math.signum(diff);
+  }
+
+  /**
+   * Calculate the shares for {@code first} and {@code second} according to
+   * {@code clusterRes}, and store the results in {@code firstShares} and
+   * {@code secondShares}, respectively. All parameters must be non-null.
+   * @param clusterRes the array of ResourceInformation instances that
+   * represents the cluster's maximum resources
+   * @param first the first resource to compare
+   * @param second the second resource to compare
+   * @param firstShares an array to store the shares for the first resource
+   * @param secondShares an array to store the shares for the second resource
+   * @return -1.0, 0.0, or 1.0, depending on whether the max share of the first
+   * resource is less than, equal to, or greater than the max share of the
+   * second resource, respectively
+   * @throws NullPointerException if any parameter is null
+   */
+  private void calculateShares(ResourceInformation[] clusterRes, Resource first,
+      Resource second, double[] firstShares, double[] secondShares) {
+    ResourceInformation[] firstRes = first.getResources();
+    ResourceInformation[] secondRes = second.getResources();
+
+    for (int i = 0; i < clusterRes.length; i++) {
+      firstShares[i] = calculateShare(clusterRes[i], firstRes[i]);
+      secondShares[i] = calculateShare(clusterRes[i], secondRes[i]);
+    }
+  }
 
-    if (l < r) {
+  /**
+   * Calculate the shares for {@code first} and {@code second} according to
+   * {@code clusterRes}, and store the results in {@code firstShares} and
+   * {@code secondShares}, respectively. All parameters must be non-null.
+   * This method assumes that the length of {@code clusterRes} is exactly 2 and
+   * makes performance optimizations based on that assumption.
+   * @param clusterRes the array of ResourceInformation instances that
+   * represents the cluster's maximum resources
+   * @param first the first resource to compare
+   * @param second the second resource to compare
+   * @param firstShares an array to store the shares for the first resource
+   * @param secondShares an array to store the shares for the second resource
+   * @return -1.0, 0.0, or 1.0, depending on whether the max share of the first
+   * resource is less than, equal to, or greater than the max share of the
+   * second resource, respectively
+   * @throws NullPointerException if any parameter is null
+   */
+  private int calculateSharesForMandatoryResources(
+      ResourceInformation[] clusterRes, Resource first, Resource second,
+      double[] firstShares, double[] secondShares) {
+    ResourceInformation[] firstRes = first.getResources();
+    ResourceInformation[] secondRes = second.getResources();
+    int firstDom = 0;
+    int secondDom = 0;
+    int firstSub = 0;
+    int secondSub = 0;
+
+    for (int i = 0; i < clusterRes.length; i++) {
+      firstShares[i] = calculateShare(clusterRes[i], firstRes[i]);
+      secondShares[i] = calculateShare(clusterRes[i], secondRes[i]);
+
+      if (firstShares[i] > firstShares[firstDom]) {
+        firstDom = i;
+      }
+
+      if (firstShares[i] < firstShares[firstSub]) {
+        firstSub = i;
+      }
+
+      if (secondShares[i] > secondShares[secondDom]) {
+        secondDom = i;
+      }
+
+      if (secondShares[i] < secondShares[secondSub]) {
+        secondSub = i;
+      }
+    }
+
+    if (firstShares[firstDom] > secondShares[secondDom]) {
+      return 1;
+    } else if (firstShares[firstDom] < secondShares[secondDom]) {
       return -1;
-    } else if (l > r) {
+    } else if (firstShares[firstSub] > secondShares[secondSub]) {
       return 1;
-    } else if (!singleType) {
-      l = getResourceAsValue(clusterResource, lhs, false);
-      r = getResourceAsValue(clusterResource, rhs, false);
-
-      if (l < r) {
-        return -1;
-      } else if (l > r) {
-        return 1;
+    } else if (firstShares[firstSub] < secondShares[secondSub]) {
+      return -1;
+    } else {
+      return 0;
+    }
+  }
+
+  /**
+   * Calculate the shares for {@code first} and {@code second} according to
+   * {@code clusterRes}, and store the results in {@code firstShares} and
+   * {@code secondShares}, respectively. {@code max} will be populated with
+   * the max shares from {@code firstShare} and {@code secondShare} in the
+   * first and second indices, respectively. All parameters must be non-null,
+   * and {@code max} must have a length of at least 2.
+   * @param clusterRes the array of ResourceInformation instances that
+   * represents the cluster's maximum resources
+   * @param first the first resource to compare
+   * @param second the second resource to compare
+   * @param firstShares an array to store the shares for the first resource
+   * @param secondShares an array to store the shares for the second resource
+   * @param max an array to store the max shares of the first and second
+   * resources
+   * @return -1.0, 0.0, or 1.0, depending on whether the max share of the first
+   * resource is less than, equal to, or greater than the max share of the
+   * second resource, respectively
+   * @throws NullPointerException if any parameter is null
+   * @throws ArrayIndexOutOfBoundsException if the length of {@code max} is
+   * less than 2
+   */
+  private void calculateShares(ResourceInformation[] clusterRes, Resource first,
+      Resource second, double[] firstShares, double[] secondShares,
+      double[] max) {
+    ResourceInformation[] firstRes = first.getResources();
+    ResourceInformation[] secondRes = second.getResources();
+
+    max[0] = 0.0;
+    max[1] = 0.0;
+
+    for (int i = 0; i < clusterRes.length; i++) {
+      firstShares[i] = calculateShare(clusterRes[i], firstRes[i]);
+      secondShares[i] = calculateShare(clusterRes[i], secondRes[i]);
+
+      if (firstShares[i] > max[0]) {
+        max[0] = firstShares[i];
+      }
+
+      if (secondShares[i] > max[1]) {
+        max[1] = secondShares[i];
       }
     }
+  }
+
+  /**
+   * Calculate the share for a resource type.
+   * @param clusterRes the resource type for the cluster maximum
+   * @param res the resource type for which to calculate the share
+   * @return the share
+   */
+  private double calculateShare(ResourceInformation clusterRes,
+      ResourceInformation res) {
+      // Convert the resources' units into the cluster resource's units
+    long value = UnitsConversionUtil.convert(res.getUnits(),
+          clusterRes.getUnits(), res.getValue());
 
-    return 0;
+    return (double) value / clusterRes.getValue();
   }
 
   /**
-   * Use 'dominant' for now since we only have 2 resources - gives us a slight
-   * performance boost.
-   * <p></p>
-   * Once we add more resources, we'll need a more complicated (and slightly
-   * less performant algorithm).
+   * Compare the two shares arrays by comparing the largest elements, then the
+   * next largest if the previous were equal, etc. The share arrays must be
+   * sorted in ascending order.
+   * @param lhsShares the first share array to compare
+   * @param rhsShares the second share array to compare
+   * @return a number that is less than 0 if the first array is less than the
+   * second, equal to 0 if the arrays are equal, and greater than 0 if the
+   * first array is greater than the second
    */
-  protected float getResourceAsValue(Resource clusterResource,
-      Resource resource, boolean dominant) {
+  private double compareShares(double[] lhsShares, double[] rhsShares) {
+    double diff = 0.0;
 
-    float min = Float.MAX_VALUE;
-    float max = 0.0f;
-    int maxLength = ResourceUtils.getResourceTypesArray().length;
-    for (int i = 0; i < maxLength; i++) {
-      ResourceInformation clusterResourceResourceInformation = clusterResource
-          .getResourceInformation(i);
-      ResourceInformation resourceInformation = resource
-          .getResourceInformation(i);
-      long resourceValue = UnitsConversionUtil.convert(
-          resourceInformation.getUnits(),
-          clusterResourceResourceInformation.getUnits(),
-          resourceInformation.getValue());
-      float tmp = (float) resourceValue
-          / (float) clusterResourceResourceInformation.getValue();
-      min = min < tmp ? min : tmp;
-      max = max > tmp ? max : tmp;
+    // lhsShares and rhsShares must necessarily have the same length, because
+    // everyone uses the same master resource list.
+    for (int i = lhsShares.length - 1; i >= 0; i--) {
+      diff = lhsShares[i] - rhsShares[i];
+
+      if (diff != 0.0) {
+        break;
+      }
     }
-    return (dominant) ? max : min;
+
+    return diff;
   }
 
   @Override
@@ -175,9 +358,19 @@ public class DominantResourceCalculator extends ResourceCalculator {
   @Override
   public float divide(Resource clusterResource,
       Resource numerator, Resource denominator) {
-    return
-        getResourceAsValue(clusterResource, numerator, true) /
-        getResourceAsValue(clusterResource, denominator, true);
+    ResourceInformation[] clusterRes = clusterResource.getResources();
+    // We have to provide the calculateShares() method with somewhere to store
+    // the shares. We don't actually need these shares afterwards.
+    double[] numeratorShares = new double[clusterRes.length];
+    double[] denominatorShares = new double[clusterRes.length];
+    // We also have to provide a place for calculateShares() to store the max
+    // shares so that we can use them.
+    double[] max = new double[2];
+
+    calculateShares(clusterRes, numerator, denominator, numeratorShares,
+        denominatorShares, max);
+
+    return (float) (max[0] / max[1]);
   }
 
   @Override

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/util/resource/TestResourceCalculator.java

@@ -21,15 +21,21 @@ package org.apache.hadoop.yarn.util.resource;
 import java.util.Arrays;
 import java.util.Collection;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.yarn.api.records.Resource;
-import org.junit.Assert;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
 @RunWith(Parameterized.class)
 public class TestResourceCalculator {
-  private ResourceCalculator resourceCalculator;
+  private final ResourceCalculator resourceCalculator;
 
   @Parameterized.Parameters
   public static Collection<ResourceCalculator[]> getParameters() {
@@ -38,6 +44,15 @@ public class TestResourceCalculator {
         { new DominantResourceCalculator() } });
   }
 
+  @BeforeClass
+  public static void setup() {
+    Configuration conf = new Configuration();
+
+    conf.set(YarnConfiguration.RESOURCE_TYPES, "test");
+    ResourceUtils.resetResourceTypes(conf);
+    ResourceUtils.getResourceTypes();
+  }
+
   public TestResourceCalculator(ResourceCalculator rs) {
     this.resourceCalculator = rs;
   }
@@ -47,32 +62,181 @@ public class TestResourceCalculator {
     Resource cluster = Resource.newInstance(1024, 1);
 
     if (resourceCalculator instanceof DefaultResourceCalculator) {
-      Assert.assertTrue(resourceCalculator.fitsIn(cluster,
+      assertTrue(resourceCalculator.fitsIn(cluster,
           Resource.newInstance(1, 2), Resource.newInstance(2, 1)));
-      Assert.assertTrue(resourceCalculator.fitsIn(cluster,
+      assertTrue(resourceCalculator.fitsIn(cluster,
           Resource.newInstance(1, 2), Resource.newInstance(2, 2)));
-      Assert.assertTrue(resourceCalculator.fitsIn(cluster,
+      assertTrue(resourceCalculator.fitsIn(cluster,
           Resource.newInstance(1, 2), Resource.newInstance(1, 2)));
-      Assert.assertTrue(resourceCalculator.fitsIn(cluster,
+      assertTrue(resourceCalculator.fitsIn(cluster,
           Resource.newInstance(1, 2), Resource.newInstance(1, 1)));
-      Assert.assertFalse(resourceCalculator.fitsIn(cluster,
+      assertFalse(resourceCalculator.fitsIn(cluster,
           Resource.newInstance(2, 1), Resource.newInstance(1, 2)));
     } else if (resourceCalculator instanceof DominantResourceCalculator) {
-      Assert.assertFalse(resourceCalculator.fitsIn(cluster,
+      assertFalse(resourceCalculator.fitsIn(cluster,
           Resource.newInstance(1, 2), Resource.newInstance(2, 1)));
-      Assert.assertTrue(resourceCalculator.fitsIn(cluster,
+      assertTrue(resourceCalculator.fitsIn(cluster,
           Resource.newInstance(1, 2), Resource.newInstance(2, 2)));
-      Assert.assertTrue(resourceCalculator.fitsIn(cluster,
+      assertTrue(resourceCalculator.fitsIn(cluster,
           Resource.newInstance(1, 2), Resource.newInstance(1, 2)));
-      Assert.assertFalse(resourceCalculator.fitsIn(cluster,
+      assertFalse(resourceCalculator.fitsIn(cluster,
           Resource.newInstance(1, 2), Resource.newInstance(1, 1)));
-      Assert.assertFalse(resourceCalculator.fitsIn(cluster,
+      assertFalse(resourceCalculator.fitsIn(cluster,
           Resource.newInstance(2, 1), Resource.newInstance(1, 2)));
     }
   }
 
+  private Resource newResource(long memory, int cpu, int test) {
+    Resource res = Resource.newInstance(memory, cpu);
+
+    res.setResourceValue("test", test);
+
+    return res;
+  }
+
+  /**
+   * Test that the compare() method returns the expected result (0, -1, or 1).
+   * If the expected result is not 0, this method will also test the resources
+   * in the opposite order and check for the negative of the expected result.
+   *
+   * @param cluster the cluster resource
+   * @param res1 the LHS resource
+   * @param res2 the RHS resource
+   * @param expected the expected result
+   */
+  private void assertComparison(Resource cluster, Resource res1, Resource res2,
+      int expected) {
+    int actual = resourceCalculator.compare(cluster, res1, res2);
+
+    assertEquals(String.format("Resource comparison did not give the expected "
+        + "result for %s v/s %s", res1.toString(), res2.toString()),
+        expected, actual);
+
+    if (expected != 0) {
+      // Try again with args in the opposite order and the negative of the
+      // expected result.
+      actual = resourceCalculator.compare(cluster, res2, res1);
+      assertEquals(String.format("Resource comparison did not give the "
+          + "expected result for %s v/s %s", res2.toString(), res1.toString()),
+          expected * -1, actual);
+    }
+  }
+
+  @Test
+  public void testCompare2() {
+    // Keep cluster resources even so that the numbers are easy to understand
+    Resource cluster = Resource.newInstance(4, 4);
+
+    assertComparison(cluster, Resource.newInstance(1, 1),
+        Resource.newInstance(1, 1), 0);
+    assertComparison(cluster, Resource.newInstance(0, 0),
+        Resource.newInstance(0, 0), 0);
+    assertComparison(cluster, Resource.newInstance(2, 2),
+        Resource.newInstance(1, 1), 1);
+    assertComparison(cluster, Resource.newInstance(2, 2),
+        Resource.newInstance(0, 0), 1);
+
+    if (resourceCalculator instanceof DefaultResourceCalculator) {
+      testCompareDefault2(cluster);
+    } else if (resourceCalculator instanceof DominantResourceCalculator) {
+      testCompareDominant2(cluster);
+    }
+  }
+
+  @Test
+  public void testCompare() {
+    // Keep cluster resources even so that the numbers are easy to understand
+    Resource cluster = newResource(4L, 4, 4);
+
+    assertComparison(cluster, newResource(1, 1, 1), newResource(1, 1, 1), 0);
+    assertComparison(cluster, newResource(0, 0, 0), newResource(0, 0, 0), 0);
+    assertComparison(cluster, newResource(2, 2, 2), newResource(1, 1, 1), 1);
+    assertComparison(cluster, newResource(2, 2, 2), newResource(0, 0, 0), 1);
+
+    if (resourceCalculator instanceof DefaultResourceCalculator) {
+      testCompareDefault(cluster);
+    } else if (resourceCalculator instanceof DominantResourceCalculator) {
+      testCompareDominant(cluster);
+    }
+  }
+
+  private void testCompareDefault2(Resource cluster) {
+    assertComparison(cluster, Resource.newInstance(1, 1),
+        Resource.newInstance(1, 1), 0);
+    assertComparison(cluster, Resource.newInstance(1, 2),
+        Resource.newInstance(1, 1), 0);
+    assertComparison(cluster, Resource.newInstance(1, 1),
+        Resource.newInstance(1, 0), 0);
+    assertComparison(cluster, Resource.newInstance(2, 1),
+        Resource.newInstance(1, 1), 1);
+    assertComparison(cluster, Resource.newInstance(2, 1),
+        Resource.newInstance(1, 2), 1);
+    assertComparison(cluster, Resource.newInstance(2, 1),
+        Resource.newInstance(1, 0), 1);
+  }
+
+  private void testCompareDominant2(Resource cluster) {
+    assertComparison(cluster, Resource.newInstance(2, 1),
+        Resource.newInstance(2, 1), 0);
+    assertComparison(cluster, Resource.newInstance(2, 1),
+        Resource.newInstance(1, 2), 0);
+    assertComparison(cluster, Resource.newInstance(2, 1),
+        Resource.newInstance(1, 1), 1);
+    assertComparison(cluster, Resource.newInstance(2, 2),
+        Resource.newInstance(2, 1), 1);
+    assertComparison(cluster, Resource.newInstance(2, 2),
+        Resource.newInstance(1, 2), 1);
+    assertComparison(cluster, Resource.newInstance(3, 1),
+        Resource.newInstance(3, 0), 1);
+  }
+
+  private void testCompareDefault(Resource cluster) {
+    assertComparison(cluster, newResource(1, 1, 2), newResource(1, 1, 1), 0);
+    assertComparison(cluster, newResource(1, 2, 1), newResource(1, 1, 1), 0);
+    assertComparison(cluster, newResource(1, 2, 2), newResource(1, 1, 1), 0);
+    assertComparison(cluster, newResource(1, 2, 2), newResource(1, 0, 0), 0);
+    assertComparison(cluster, newResource(2, 1, 1), newResource(1, 1, 1), 1);
+    assertComparison(cluster, newResource(2, 1, 1), newResource(1, 2, 1), 1);
+    assertComparison(cluster, newResource(2, 1, 1), newResource(1, 1, 2), 1);
+    assertComparison(cluster, newResource(2, 1, 1), newResource(1, 2, 2), 1);
+    assertComparison(cluster, newResource(2, 1, 1), newResource(1, 0, 0), 1);
+  }
+
+  private void testCompareDominant(Resource cluster) {
+    assertComparison(cluster, newResource(2, 1, 1), newResource(2, 1, 1), 0);
+    assertComparison(cluster, newResource(2, 1, 1), newResource(1, 2, 1), 0);
+    assertComparison(cluster, newResource(2, 1, 1), newResource(1, 1, 2), 0);
+    assertComparison(cluster, newResource(2, 1, 0), newResource(0, 1, 2), 0);
+    assertComparison(cluster, newResource(2, 2, 1), newResource(1, 2, 2), 0);
+    assertComparison(cluster, newResource(2, 2, 1), newResource(2, 1, 2), 0);
+    assertComparison(cluster, newResource(2, 2, 1), newResource(2, 2, 1), 0);
+    assertComparison(cluster, newResource(2, 2, 0), newResource(2, 0, 2), 0);
+    assertComparison(cluster, newResource(3, 2, 1), newResource(3, 2, 1), 0);
+    assertComparison(cluster, newResource(3, 2, 1), newResource(3, 1, 2), 0);
+    assertComparison(cluster, newResource(3, 2, 1), newResource(1, 2, 3), 0);
+    assertComparison(cluster, newResource(3, 2, 1), newResource(1, 3, 2), 0);
+    assertComparison(cluster, newResource(3, 2, 1), newResource(2, 1, 3), 0);
+    assertComparison(cluster, newResource(3, 2, 1), newResource(2, 3, 1), 0);
+    assertComparison(cluster, newResource(2, 1, 1), newResource(1, 1, 1), 1);
+    assertComparison(cluster, newResource(2, 1, 1), newResource(1, 1, 0), 1);
+    assertComparison(cluster, newResource(2, 2, 1), newResource(2, 1, 1), 1);
+    assertComparison(cluster, newResource(2, 2, 1), newResource(1, 2, 1), 1);
+    assertComparison(cluster, newResource(2, 2, 1), newResource(1, 1, 2), 1);
+    assertComparison(cluster, newResource(2, 2, 1), newResource(0, 2, 2), 1);
+    assertComparison(cluster, newResource(2, 2, 2), newResource(2, 1, 1), 1);
+    assertComparison(cluster, newResource(2, 2, 2), newResource(1, 2, 1), 1);
+    assertComparison(cluster, newResource(2, 2, 2), newResource(1, 1, 2), 1);
+    assertComparison(cluster, newResource(2, 2, 2), newResource(2, 2, 1), 1);
+    assertComparison(cluster, newResource(2, 2, 2), newResource(2, 1, 2), 1);
+    assertComparison(cluster, newResource(2, 2, 2), newResource(1, 2, 2), 1);
+    assertComparison(cluster, newResource(3, 2, 1), newResource(2, 2, 2), 1);
+    assertComparison(cluster, newResource(3, 1, 1), newResource(2, 2, 2), 1);
+    assertComparison(cluster, newResource(3, 1, 1), newResource(3, 1, 0), 1);
+    assertComparison(cluster, newResource(3, 1, 1), newResource(3, 0, 0), 1);
+  }
+
   @Test(timeout = 10000)
-  public void testResourceCalculatorCompareMethod() {
+  public void testCompareWithEmptyCluster() {
     Resource clusterResource = Resource.newInstance(0, 0);
 
     // For lhs == rhs
@@ -126,27 +290,27 @@ public class TestResourceCalculator {
       boolean greaterThan, boolean greaterThanOrEqual, Resource max,
       Resource min) {
 
-    Assert.assertEquals("Less Than operation is wrongly calculated.", lessThan,
+    assertEquals("Less Than operation is wrongly calculated.", lessThan,
         Resources.lessThan(resourceCalculator, clusterResource, lhs, rhs));
 
-    Assert.assertEquals(
+    assertEquals(
         "Less Than Or Equal To operation is wrongly calculated.",
         lessThanOrEqual, Resources.lessThanOrEqual(resourceCalculator,
             clusterResource, lhs, rhs));
 
-    Assert.assertEquals("Greater Than operation is wrongly calculated.",
+    assertEquals("Greater Than operation is wrongly calculated.",
         greaterThan,
         Resources.greaterThan(resourceCalculator, clusterResource, lhs, rhs));
 
-    Assert.assertEquals(
+    assertEquals(
         "Greater Than Or Equal To operation is wrongly calculated.",
         greaterThanOrEqual, Resources.greaterThanOrEqual(resourceCalculator,
             clusterResource, lhs, rhs));
 
-    Assert.assertEquals("Max(value) Operation wrongly calculated.", max,
+    assertEquals("Max(value) Operation wrongly calculated.", max,
         Resources.max(resourceCalculator, clusterResource, lhs, rhs));
 
-    Assert.assertEquals("Min(value) operation is wrongly calculated.", min,
+    assertEquals("Min(value) operation is wrongly calculated.", min,
         Resources.min(resourceCalculator, clusterResource, lhs, rhs));
   }
 
@@ -164,13 +328,13 @@ public class TestResourceCalculator {
       Resource result = Resources.normalize(resourceCalculator,
           ask, min, max, increment);
 
-      Assert.assertEquals(2 * 1024, result.getMemorySize());
+      assertEquals(2 * 1024, result.getMemorySize());
     } else if (resourceCalculator instanceof DominantResourceCalculator) {
       Resource result = Resources.normalize(resourceCalculator,
           ask, min, max, increment);
 
-      Assert.assertEquals(2 * 1024, result.getMemorySize());
-      Assert.assertEquals(4, result.getVirtualCores());
+      assertEquals(2 * 1024, result.getMemorySize());
+      assertEquals(4, result.getVirtualCores());
     }
 
     // if resources asked are less than minimum resource, then normalize it to
@@ -183,13 +347,13 @@ public class TestResourceCalculator {
       Resource result = Resources.normalize(resourceCalculator,
           ask, min, max, increment);
 
-      Assert.assertEquals(2 * 1024, result.getMemorySize());
+      assertEquals(2 * 1024, result.getMemorySize());
     } else if (resourceCalculator instanceof DominantResourceCalculator) {
       Resource result = Resources.normalize(resourceCalculator,
           ask, min, max, increment);
 
-      Assert.assertEquals(2 * 1024, result.getMemorySize());
-      Assert.assertEquals(2, result.getVirtualCores());
+      assertEquals(2 * 1024, result.getMemorySize());
+      assertEquals(2, result.getVirtualCores());
     }
 
     // if resources asked are larger than maximum resource, then normalize it to
@@ -202,13 +366,13 @@ public class TestResourceCalculator {
       Resource result = Resources.normalize(resourceCalculator,
           ask, min, max, increment);
 
-      Assert.assertEquals(8 * 1024, result.getMemorySize());
+      assertEquals(8 * 1024, result.getMemorySize());
     } else if (resourceCalculator instanceof DominantResourceCalculator) {
       Resource result = Resources.normalize(resourceCalculator,
           ask, min, max, increment);
 
-      Assert.assertEquals(8 * 1024, result.getMemorySize());
-      Assert.assertEquals(8, result.getVirtualCores());
+      assertEquals(8 * 1024, result.getMemorySize());
+      assertEquals(8, result.getVirtualCores());
     }
 
     // if increment is 0, use minimum resource as the increment resource.
@@ -220,13 +384,13 @@ public class TestResourceCalculator {
       Resource result = Resources.normalize(resourceCalculator,
           ask, min, max, increment);
 
-      Assert.assertEquals(2 * 1024, result.getMemorySize());
+      assertEquals(2 * 1024, result.getMemorySize());
     } else if (resourceCalculator instanceof DominantResourceCalculator) {
       Resource result = Resources.normalize(resourceCalculator,
           ask, min, max, increment);
 
-      Assert.assertEquals(2 * 1024, result.getMemorySize());
-      Assert.assertEquals(2, result.getVirtualCores());
+      assertEquals(2 * 1024, result.getMemorySize());
+      assertEquals(2, result.getVirtualCores());
     }
   }
 }