diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationFileLoaderService.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationFileLoaderService.java index 661caa7..2a1644b 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationFileLoaderService.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationFileLoaderService.java @@ -42,6 +42,8 @@ import org.apache.hadoop.yarn.api.records.ReservationACL; import org.apache.hadoop.yarn.api.records.Resource; import org.apache.hadoop.yarn.server.resourcemanager.resource.ResourceWeights; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.DominantResourceFairnessPolicy; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FairSharePolicy; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FifoPolicy; import org.apache.hadoop.yarn.util.Clock; import org.apache.hadoop.yarn.util.SystemClock; @@ -237,6 +239,7 @@ public synchronized void reloadAllocations() throws IOException, long defaultMinSharePreemptionTimeout = Long.MAX_VALUE; float defaultFairSharePreemptionThreshold = 0.5f; SchedulingPolicy defaultSchedPolicy = SchedulingPolicy.DEFAULT_POLICY; + Map parents = new HashMap(); // Reservation global configuration knobs String planner = null; @@ -370,7 +373,8 @@ public synchronized void reloadAllocations() throws IOException, queueMaxApps, userMaxApps, queueMaxAMShares, queueWeights, queuePolicies, minSharePreemptionTimeouts, fairSharePreemptionTimeouts, fairSharePreemptionThresholds, queueAcls, reservationAcls, - configuredQueues, reservableQueues, nonPreemptableQueues); + configuredQueues, reservableQueues, nonPreemptableQueues, + parents, defaultSchedPolicy); } // Load placement policy and pass it configured queues @@ -443,9 +447,12 @@ private void loadQueue(String parentName, Element element, Map> resAcls, Map> configuredQueues, Set reservableQueues, - Set nonPreemptableQueues) + Set nonPreemptableQueues, + Map parents, + SchedulingPolicy defaultSchedPolicy) throws AllocationConfigurationException { String queueName = element.getAttribute("name").trim(); + parents.put(queueName, parentName); if (queueName.contains(".")) { throw new AllocationConfigurationException("Bad fair scheduler config " @@ -510,6 +517,8 @@ private void loadQueue(String parentName, Element element, || "schedulingMode".equals(field.getTagName())) { String text = ((Text)field.getFirstChild()).getData().trim(); SchedulingPolicy policy = SchedulingPolicy.parse(text); + checkIfParentPoliceAllow(policy, parentName, queuePolicies, parents, + defaultSchedPolicy); queuePolicies.put(queueName, policy); } else if ("aclSubmitApps".equals(field.getTagName())) { String text = ((Text)field.getFirstChild()).getData(); @@ -545,7 +554,7 @@ private void loadQueue(String parentName, Element element, queuePolicies, minSharePreemptionTimeouts, fairSharePreemptionTimeouts, fairSharePreemptionThresholds, queueAcls, resAcls, configuredQueues, reservableQueues, - nonPreemptableQueues); + nonPreemptableQueues, parents, defaultSchedPolicy); isLeaf = false; } } @@ -578,7 +587,68 @@ private void loadQueue(String parentName, Element element, minQueueResources.get(queueName))); } } - + + + /** + * Recursive check its parent policy is allowed. + * @param policy the policy of current queue + * @param parentName the name of parent queue + * @param queuePolicies a map of queue and its policy + * @param parents a map of queue and the name of its parent + * @throws AllocationConfigurationException if parent policy is not allowed. + */ + private void checkIfParentPoliceAllow(SchedulingPolicy policy, + String parentName, Map queuePolicies, + Map parents, SchedulingPolicy defaultSchedPolicy) + throws AllocationConfigurationException { + if (parentName == null) { + return; + } + + Set policies = allowParentPolicies(policy); + SchedulingPolicy parentPolicy = queuePolicies.get(parentName); + if (parentPolicy == null) { + parentPolicy = defaultSchedPolicy; + } + + if (policies.contains(parentPolicy.getName())) { + checkIfParentPoliceAllow(policy, parents.get(parentName), + queuePolicies, parents, defaultSchedPolicy); + } else { + StringBuffer validPoliciesBuf = new StringBuffer(); + int i = 0; + for (String item : policies) { + i++; + if (i == policies.size()) { + validPoliciesBuf.append(item); + } else { + validPoliciesBuf.append(item + ", "); + } + } + + throw new AllocationConfigurationException("Parent scheduling policy " + + "can only be in (" + validPoliciesBuf.toString() + ") if the " + + "scheduling policy of current queue is" + policy.getName()); + } + } + + /** + * Return all allowed parent policies based on the current queue policy. + */ + private Set allowParentPolicies(SchedulingPolicy policy) { + Set policies = new HashSet<>(); + if (policy.getName().equals(FifoPolicy.NAME)) { + policies.add(FairSharePolicy.NAME); + policies.add(DominantResourceFairnessPolicy.NAME); + } else if (policy.getName().equals(FairSharePolicy.NAME)) { + policies.add(FairSharePolicy.NAME); + policies.add(DominantResourceFairnessPolicy.NAME); + } else if (policy.getName().equals(DominantResourceFairnessPolicy.NAME)) { + policies.add(DominantResourceFairnessPolicy.NAME); + } + return policies; + } + public interface Listener { public void onReload(AllocationConfiguration info); } diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestAllocationFileLoaderService.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestAllocationFileLoaderService.java index cc91ef9..1b4223f 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestAllocationFileLoaderService.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestAllocationFileLoaderService.java @@ -557,6 +557,33 @@ public void testQueueNameContainingPeriods() throws Exception { } /** + * Verify whether scheduling policy of parent queue is allowed. + */ + @Test (expected = AllocationConfigurationException.class) + public void testParentPolicyAllowed() throws Exception { + Configuration conf = new Configuration(); + conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, ALLOC_FILE); + + PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE)); + out.println(""); + out.println(""); + out.println(""); + out.println("fair"); + out.println(" "); + out.println(" drf"); + out.println(" "); + out.println(""); + out.println(""); + out.close(); + + AllocationFileLoaderService allocLoader = new AllocationFileLoaderService(); + allocLoader.init(conf); + ReloadListener confHolder = new ReloadListener(); + allocLoader.setReloadListener(confHolder); + allocLoader.reloadAllocations(); + } + + /** * Verify that you can't have the queue name with whitespace only in the * allocations file. */