Added missing handling of operator subtractTestNonNegative

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/RLESparseResourceAllocation.java

@@ -440,8 +440,18 @@ public class RLESparseResourceAllocation {
       if (eB == null || eB.getValue() == null) {
         return null;
       }
-      if (op == RLEOperator.subtract) {
+      if (op == RLEOperator.subtract
-        return Resources.negate(eB.getValue());
+          || op == RLEOperator.subtractTestNonNegative) {
+        Resource val = Resources.negate(eB.getValue());
+        // test for negative value and throws
+        if (op == RLEOperator.subtractTestNonNegative
+            && (Resources.fitsIn(val, ZERO_RESOURCE)
+                && !Resources.equals(val, ZERO_RESOURCE))) {
+          throw new PlanningException(
+              "RLESparseResourceAllocation: merge failed as the "
+                  + "resulting RLESparseResourceAllocation would be negative");
+        }
+        return val;
       } else {
         return eB.getValue();
       }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestRLESparseResourceAllocation.java

@@ -236,6 +236,23 @@ public class TestRLESparseResourceAllocation {
       // Expected!
     }
 
+    // try with null value of an entry in a
+    a.put(10L, null);
+    b.put(11L, Resource.newInstance(10, 5));
+
+    rleA = new RLESparseResourceAllocation(a, new DefaultResourceCalculator());
+    rleB = new RLESparseResourceAllocation(b, new DefaultResourceCalculator());
+
+    try {
+      RLESparseResourceAllocation out =
+          RLESparseResourceAllocation.merge(new DefaultResourceCalculator(),
+              Resource.newInstance(100 * 128 * 1024, 100 * 32), rleA, rleB,
+              RLEOperator.subtractTestNonNegative, 0, 60);
+      fail();
+    } catch (PlanningException pe) {
+      // Expected!
+    }
+
     // trying a case that should work
     a.put(10L, Resource.newInstance(10, 6));
     b.put(11L, Resource.newInstance(5, 6));