diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/placement/AppNameMappingPlacementRule.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/placement/AppNameMappingPlacementRule.java new file mode 100644 index 00000000000..3d002287c65 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/placement/AppNameMappingPlacementRule.java @@ -0,0 +1,241 @@ +package org.apache.hadoop.yarn.server.resourcemanager.placement; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext; +import org.apache.hadoop.yarn.exceptions.YarnException; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueue; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerContext; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerQueueManager; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.LeafQueue; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.ManagedParentQueue; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.AutoCreatedLeafQueue; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration.DOT; + +public class AppNameMappingPlacementRule extends PlacementRule{ + private static final Log LOG = LogFactory + .getLog(AppNameMappingPlacementRule.class); + + public static final String CURRENT_USER_MAPPING = "%user"; + + public static final String PRIMARY_GROUP_MAPPING = "%primary_group"; + + private static final String QUEUE_MAPPING_NAME = "app-name-mapping-rule"; + + private boolean overrideWithQueueMappings = false; + private List mappings = null; + + + private static class QueuePath { + + public String parentQueue; + public String leafQueue; + + public QueuePath(final String leafQueue) { + this.leafQueue = leafQueue; + } + + public QueuePath(final String parentQueue, final String leafQueue) { + this.parentQueue = parentQueue; + this.leafQueue = leafQueue; + } + + public String getParentQueue() { + return parentQueue; + } + + public String getLeafQueue() { + return leafQueue; + } + + public boolean hasParentQueue() { + return parentQueue != null; + } + + @Override + public String toString() { + return parentQueue + DOT + leafQueue; + } + } + + private static QueuePath extractQueuePath(String queueName) + throws IOException { + int parentQueueNameEndIndex = queueName.lastIndexOf(DOT); + + if (parentQueueNameEndIndex > -1) { + final String parentQueue = queueName.substring(0, parentQueueNameEndIndex) + .trim(); + final String leafQueue = queueName.substring(parentQueueNameEndIndex + 1) + .trim(); + return new QueuePath(parentQueue, leafQueue); + } + + return new QueuePath(queueName); + } + + private static boolean isStaticQueueMapping(QueueMappingEntity mapping) { + return !mapping.getQueue().contains(CURRENT_USER_MAPPING) && !mapping + .getQueue().contains(PRIMARY_GROUP_MAPPING); + } + + private static boolean ifQueueDoesNotExist(CSQueue queue) { + return queue == null; + } + + private static void validateParentQueue(CSQueue parentQueue, + String parentQueueName, String leafQueueName) throws IOException { + if (parentQueue == null) { + throw new IOException( + "mapping contains invalid or non-leaf queue [" + leafQueueName + + "] and invalid parent queue [" + parentQueueName + "]"); + } else if (!(parentQueue instanceof ManagedParentQueue)) { + throw new IOException("mapping contains leaf queue [" + leafQueueName + + "] and invalid parent queue which " + + "does not have auto creation of leaf queues enabled [" + + parentQueueName + "]"); + } else if (!parentQueue.getQueueName().equals(parentQueueName)) { + throw new IOException( + "mapping contains invalid or non-leaf queue [" + leafQueueName + + "] and invalid parent queue " + + "which does not match existing leaf queue's parent : [" + + parentQueueName + "] does not match [ " + parentQueue + .getQueueName() + "]"); + } + } + + private static QueueMappingEntity validateAndGetAutoCreatedQueueMapping( + CapacitySchedulerQueueManager queueManager, QueueMappingEntity mapping, + QueuePath queuePath) throws IOException { + if (queuePath.hasParentQueue()) { + //if parent queue is specified, + // then it should exist and be an instance of ManagedParentQueue + validateParentQueue(queueManager.getQueue(queuePath.getParentQueue()), + queuePath.getParentQueue(), queuePath.getLeafQueue()); + return new QueueMappingEntity(mapping.getSource(), + queuePath.getLeafQueue(), queuePath.getParentQueue()); + } + + return null; + } + + private static QueueMappingEntity validateAndGetQueueMapping( + CapacitySchedulerQueueManager queueManager, CSQueue queue, + QueueMappingEntity mapping, QueuePath queuePath) throws IOException { + if (!(queue instanceof LeafQueue)) { + throw new IOException( + "mapping contains invalid or non-leaf queue : " + mapping.getQueue()); + } + + if (queue instanceof AutoCreatedLeafQueue && queue + .getParent() instanceof ManagedParentQueue) { + + QueueMappingEntity newMapping = validateAndGetAutoCreatedQueueMapping( + queueManager, mapping, queuePath); + if (newMapping == null) { + throw new IOException( + "mapping contains invalid or non-leaf queue " + mapping.getQueue()); + } + return newMapping; + } + return mapping; + } + + + @Override + public void initialize(CapacitySchedulerContext schedulerContext) + throws IOException { + CapacitySchedulerConfiguration conf = schedulerContext.getConfiguration(); + boolean overrideWithQueueMappings = conf.getOverrideWithQueueMappings(); + LOG.info( + "Initialized queue mappings, override: " + overrideWithQueueMappings); + + List queueMappings = conf.getQueueMappingEntity(QUEUE_MAPPING_NAME); + + // Get new user mappings + List newMappings = new ArrayList<>(); + + CapacitySchedulerQueueManager queueManager = + schedulerContext.getCapacitySchedulerQueueManager(); + + // check if mappings refer to valid queues + for (QueueMappingEntity mapping : queueMappings) { + + QueuePath queuePath = extractQueuePath(mapping.getQueue()); + if (isStaticQueueMapping(mapping)) { + //Try getting queue by its leaf queue name + // without splitting into parent/leaf queues + CSQueue queue = queueManager.getQueue(mapping.getQueue()); + if (ifQueueDoesNotExist(queue)) { + //Try getting the queue by extracting leaf and parent queue names + //Assuming its a potential auto created leaf queue + queue = queueManager.getQueue(queuePath.getLeafQueue()); + + if (ifQueueDoesNotExist(queue)) { + //if leaf queue does not exist, + // this could be a potential auto created leaf queue + //validate if parent queue is specified, + // then it should exist and + // be an instance of AutoCreateEnabledParentQueue + QueueMappingEntity newMapping = validateAndGetAutoCreatedQueueMapping( + queueManager, mapping, queuePath); + if (newMapping == null) { + throw new IOException( + "mapping contains invalid or non-leaf queue " + mapping + .getQueue()); + } + newMappings.add(newMapping); + } else{ + QueueMappingEntity newMapping = validateAndGetQueueMapping(queueManager, + queue, mapping, queuePath); + newMappings.add(newMapping); + } + } else{ + // if queue exists, validate + // if its an instance of leaf queue + // if its an instance of auto created leaf queue, + // then extract parent queue name and update queue mapping + QueueMappingEntity newMapping = validateAndGetQueueMapping(queueManager, + queue, mapping, queuePath); + newMappings.add(newMapping); + } + } else{ + //If it is a dynamic queue mapping, + // we can safely assume leaf queue name does not have '.' in it + // validate + // if parent queue is specified, then + // parent queue exists and an instance of AutoCreateEnabledParentQueue + // + QueueMappingEntity newMapping = validateAndGetAutoCreatedQueueMapping( + queueManager, mapping, queuePath); + if (newMapping != null) { + newMappings.add(newMapping); + } else{ + newMappings.add(mapping); + } + } + } + + // initialize groups if mappings are present + if (newMappings.size() > 0) { + this.mappings = newMappings; + this.overrideWithQueueMappings = overrideWithQueueMappings; + } else { + this.mappings = null; + } + } + + @Override + public ApplicationPlacementContext getPlacementForApp( + ApplicationSubmissionContext asc, String user) throws YarnException { + return null; + } + + @Override + public List getQueueMappingLists() { return mappings; } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/placement/PlacementRule.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/placement/PlacementRule.java index a9d5e3337e5..24d5f52d96c 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/placement/PlacementRule.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/placement/PlacementRule.java @@ -18,21 +18,73 @@ package org.apache.hadoop.yarn.server.resourcemanager.placement; +import java.io.IOException; +import java.util.List; import java.util.Map; import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext; import org.apache.hadoop.yarn.exceptions.YarnException; -import org.apache.hadoop.yarn.server.resourcemanager.RMContext; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerContext; public abstract class PlacementRule { + public static class QueueMappingEntity { + String source; + String queue; + String parentQueue; + + public final static String DELIMITER = ":"; + + public QueueMappingEntity(String source, String queue) { + this.source = source; + this.queue = queue; + this.parentQueue = null; + } + public QueueMappingEntity(String source, String queue, String parentQueue) { + this.source = source; + this.queue = queue; + this.parentQueue = parentQueue; + } + + public String getQueue() { return queue; } + + public String getParentQueue() { + return parentQueue; + } + + public String getSource() { + return source; + } + + @Override + public int hashCode() { + return super.hashCode(); + } + + @Override + public boolean equals(Object obj) { + if (obj instanceof QueueMappingEntity) { + QueueMappingEntity other = (QueueMappingEntity) obj; + return (other.source.equals(source) && + other.queue.equals(queue)); + } else { + return false; + } + } + + public String toString() { + return source + DELIMITER + (parentQueue != null ? + parentQueue + "." + queue : + queue); + } + } + public String getName() { return this.getClass().getName(); } - public void initialize(Map parameters, RMContext rmContext) - throws YarnException { - } + public abstract void initialize( + CapacitySchedulerContext schedulerContext) throws IOException; /** * Get queue for a given application @@ -53,4 +105,6 @@ public void initialize(Map parameters, RMContext rmContext) */ public abstract ApplicationPlacementContext getPlacementForApp( ApplicationSubmissionContext asc, String user) throws YarnException; + + public abstract List getQueueMappingLists(); } \ No newline at end of file diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/placement/UserGroupMappingPlacementRule.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/placement/UserGroupMappingPlacementRule.java index d03b832ad8b..07efbc1a45e 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/placement/UserGroupMappingPlacementRule.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/placement/UserGroupMappingPlacementRule.java @@ -137,6 +137,8 @@ public String toString() { } } + public UserGroupMappingPlacementRule(){} + public UserGroupMappingPlacementRule(boolean overrideWithQueueMappings, List newMappings, Groups groups) { this.mappings = newMappings; @@ -217,90 +219,6 @@ private ApplicationPlacementContext getPlacementContext(QueueMapping mapping, } } - @VisibleForTesting - public static UserGroupMappingPlacementRule get( - CapacitySchedulerContext schedulerContext) throws IOException { - CapacitySchedulerConfiguration conf = schedulerContext.getConfiguration(); - boolean overrideWithQueueMappings = conf.getOverrideWithQueueMappings(); - LOG.info( - "Initialized queue mappings, override: " + overrideWithQueueMappings); - - List queueMappings = conf.getQueueMappings(); - - // Get new user/group mappings - List newMappings = new ArrayList<>(); - - CapacitySchedulerQueueManager queueManager = - schedulerContext.getCapacitySchedulerQueueManager(); - - // check if mappings refer to valid queues - for (QueueMapping mapping : queueMappings) { - - QueuePath queuePath = extractQueuePath(mapping.getQueue()); - if (isStaticQueueMapping(mapping)) { - //Try getting queue by its leaf queue name - // without splitting into parent/leaf queues - CSQueue queue = queueManager.getQueue(mapping.getQueue()); - if (ifQueueDoesNotExist(queue)) { - //Try getting the queue by extracting leaf and parent queue names - //Assuming its a potential auto created leaf queue - queue = queueManager.getQueue(queuePath.getLeafQueue()); - - if (ifQueueDoesNotExist(queue)) { - //if leaf queue does not exist, - // this could be a potential auto created leaf queue - //validate if parent queue is specified, - // then it should exist and - // be an instance of AutoCreateEnabledParentQueue - QueueMapping newMapping = validateAndGetAutoCreatedQueueMapping( - queueManager, mapping, queuePath); - if (newMapping == null) { - throw new IOException( - "mapping contains invalid or non-leaf queue " + mapping - .getQueue()); - } - newMappings.add(newMapping); - } else{ - QueueMapping newMapping = validateAndGetQueueMapping(queueManager, - queue, mapping, queuePath); - newMappings.add(newMapping); - } - } else{ - // if queue exists, validate - // if its an instance of leaf queue - // if its an instance of auto created leaf queue, - // then extract parent queue name and update queue mapping - QueueMapping newMapping = validateAndGetQueueMapping(queueManager, - queue, mapping, queuePath); - newMappings.add(newMapping); - } - } else{ - //If it is a dynamic queue mapping, - // we can safely assume leaf queue name does not have '.' in it - // validate - // if parent queue is specified, then - // parent queue exists and an instance of AutoCreateEnabledParentQueue - // - QueueMapping newMapping = validateAndGetAutoCreatedQueueMapping( - queueManager, mapping, queuePath); - if (newMapping != null) { - newMappings.add(newMapping); - } else{ - newMappings.add(mapping); - } - } - } - - // initialize groups if mappings are present - if (newMappings.size() > 0) { - Groups groups = new Groups(conf); - return new UserGroupMappingPlacementRule(overrideWithQueueMappings, - newMappings, groups); - } - - return null; - } - private static QueueMapping validateAndGetQueueMapping( CapacitySchedulerQueueManager queueManager, CSQueue queue, QueueMapping mapping, QueuePath queuePath) throws IOException { @@ -421,4 +339,98 @@ private static void validateParentQueue(CSQueue parentQueue, public List getQueueMappings() { return mappings; } + + @Override + public void initialize(CapacitySchedulerContext schedulerContext) + throws IOException { + CapacitySchedulerConfiguration conf = schedulerContext.getConfiguration(); + boolean overrideWithQueueMappings = conf.getOverrideWithQueueMappings(); + LOG.info( + "Initialized queue mappings, override: " + overrideWithQueueMappings); + + List queueMappings = conf.getQueueMappings(); + + // Get new user/group mappings + List newMappings = new ArrayList<>(); + + CapacitySchedulerQueueManager queueManager = + schedulerContext.getCapacitySchedulerQueueManager(); + + // check if mappings refer to valid queues + for (QueueMapping mapping : queueMappings) { + + QueuePath queuePath = extractQueuePath(mapping.getQueue()); + if (isStaticQueueMapping(mapping)) { + //Try getting queue by its leaf queue name + // without splitting into parent/leaf queues + CSQueue queue = queueManager.getQueue(mapping.getQueue()); + if (ifQueueDoesNotExist(queue)) { + //Try getting the queue by extracting leaf and parent queue names + //Assuming its a potential auto created leaf queue + queue = queueManager.getQueue(queuePath.getLeafQueue()); + + if (ifQueueDoesNotExist(queue)) { + //if leaf queue does not exist, + // this could be a potential auto created leaf queue + //validate if parent queue is specified, + // then it should exist and + // be an instance of AutoCreateEnabledParentQueue + QueueMapping newMapping = validateAndGetAutoCreatedQueueMapping( + queueManager, mapping, queuePath); + if (newMapping == null) { + throw new IOException( + "mapping contains invalid or non-leaf queue " + mapping + .getQueue()); + } + newMappings.add(newMapping); + } else{ + QueueMapping newMapping = validateAndGetQueueMapping(queueManager, + queue, mapping, queuePath); + newMappings.add(newMapping); + } + } else{ + // if queue exists, validate + // if its an instance of leaf queue + // if its an instance of auto created leaf queue, + // then extract parent queue name and update queue mapping + QueueMapping newMapping = validateAndGetQueueMapping(queueManager, + queue, mapping, queuePath); + newMappings.add(newMapping); + } + } else{ + //If it is a dynamic queue mapping, + // we can safely assume leaf queue name does not have '.' in it + // validate + // if parent queue is specified, then + // parent queue exists and an instance of AutoCreateEnabledParentQueue + // + QueueMapping newMapping = validateAndGetAutoCreatedQueueMapping( + queueManager, mapping, queuePath); + if (newMapping != null) { + newMappings.add(newMapping); + } else{ + newMappings.add(mapping); + } + } + } + + // initialize groups if mappings are present + if (newMappings.size() > 0) { + Groups groups = new Groups(conf); + this.mappings = newMappings; + this.groups = groups; + this.overrideWithQueueMappings = overrideWithQueueMappings; + } + } + + @Override + public List getQueueMappingLists() { + List mappingEntities= new ArrayList<>(); + for (QueueMapping mapping : mappings) { + QueueMappingEntity qme = new QueueMappingEntity(mapping.source, + mapping.queue, mapping.parentQueue); + mappingEntities.add(qme); + } + return mappingEntities; + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java index ddab0c1bd82..2aeea64f498 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java @@ -641,7 +641,9 @@ public int getPendingBacklogs() { public PlacementRule getUserGroupMappingPlacementRule() throws IOException { try { readLock.lock(); - return UserGroupMappingPlacementRule.get(this); + UserGroupMappingPlacementRule ugRule = new UserGroupMappingPlacementRule(); + ugRule.initialize(this); + return ugRule; } finally { readLock.unlock(); } @@ -672,7 +674,15 @@ void updatePlacementRules() throws IOException { PlacementRule rule = PlacementFactory.getPlacementRule( placementRuleStr, conf); if (null != rule) { - placementRules.add(rule); + try { + readLock.lock(); + rule.initialize(this); + } finally { + readLock.unlock(); + } + if (null != rule.getQueueMappingLists()) { + placementRules.add(rule); + } } } catch (ClassNotFoundException cnfe) { throw new IOException(cnfe); diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java index bdd30b915c9..b8fea0efce1 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java @@ -38,6 +38,7 @@ import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager; import org.apache.hadoop.yarn.security.AccessType; import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; +import org.apache.hadoop.yarn.server.resourcemanager.placement.PlacementRule.QueueMappingEntity; import org.apache.hadoop.yarn.server.resourcemanager.placement.UserGroupMappingPlacementRule.QueueMapping; import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSchedulerConfiguration; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; @@ -1050,6 +1051,36 @@ public void setOverrideWithQueueMappings(boolean overrideWithQueueMappings) { return mappings; } + public List getQueueMappingEntity(String queueMappingSuffix) { + String queueMappingName = buildQueueMappingRuleProperty(queueMappingSuffix); + + List mappings = + new ArrayList(); + Collection mappingsString = + getTrimmedStringCollection(queueMappingName); + for (String mappingValue : mappingsString) { + String[] mapping = + getTrimmedStringCollection(mappingValue, ":") + .toArray(new String[] {}); + if (mapping.length != 2 || mapping[1].length() == 0) { + throw new IllegalArgumentException( + "Illegal queue mapping " + mappingValue); + } + + QueueMappingEntity m = new QueueMappingEntity(mapping[0], mapping[1]); + + mappings.add(m); + } + + return mappings; + } + + private String buildQueueMappingRuleProperty (String queueMappingSuffix) { + StringBuilder queueMapping = new StringBuilder(); + queueMapping.append(QUEUE_MAPPING).append(".").append(queueMappingSuffix); + return queueMapping.toString(); + } + @Private @VisibleForTesting public void setQueuePlacementRules(Collection queuePlacementRules) { @@ -1075,6 +1106,23 @@ public void setQueueMappings(List queueMappings) { setStrings(QUEUE_MAPPING, StringUtils.join(",", queueMappingStrs)); } + @Private + @VisibleForTesting + public void setQueueMappingEntities(List queueMappings, + String queueMappingSuffix) { + if (queueMappings == null) { + return; + } + + List queueMappingStrs = new ArrayList<>(); + for (QueueMappingEntity mapping : queueMappings) { + queueMappingStrs.add(mapping.toString()); + } + + String mappingRuleProp = buildQueueMappingRuleProperty(queueMappingSuffix); + setStrings(mappingRuleProp, StringUtils.join(",", queueMappingStrs)); + } + public boolean isReservable(String queue) { boolean isReservable = getBoolean(getQueuePrefix(queue) + IS_RESERVABLE, false); diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacitySchedulerAutoCreatedQueueBase.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacitySchedulerAutoCreatedQueueBase.java index d8d71c71ea3..b1d917ebb48 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacitySchedulerAutoCreatedQueueBase.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacitySchedulerAutoCreatedQueueBase.java @@ -38,6 +38,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; import org.apache.hadoop.yarn.server.resourcemanager.placement .ApplicationPlacementContext; +import org.apache.hadoop.yarn.server.resourcemanager.placement.PlacementRule; import org.apache.hadoop.yarn.server.resourcemanager.placement .UserGroupMappingPlacementRule; import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; @@ -63,6 +64,7 @@ import org.junit.After; import org.junit.Assert; import org.junit.Before; +import org.junit.Test; import java.io.IOException; import java.util.ArrayList; @@ -83,8 +85,10 @@ .capacity.CapacitySchedulerConfiguration.DOT; import static org.apache.hadoop.yarn.server.resourcemanager.scheduler .capacity.CapacitySchedulerConfiguration.FAIR_APP_ORDERING_POLICY; +import static org.hamcrest.CoreMatchers.hasItems; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -147,6 +151,9 @@ private final TestCapacityScheduler tcs = new TestCapacityScheduler(); protected SpyDispatcher dispatcher; private static EventHandler rmAppEventEventHandler; + private static final String QUEUE_MAPPING_NAME = "app-name-mapping-rule"; + private static final String QUEUE_MAPPING_RULE_APP_NAME = + "org.apache.hadoop.yarn.server.resourcemanager.placement.AppNameMappingPlacementRule"; public static class SpyDispatcher extends AsyncDispatcher { @@ -256,6 +263,35 @@ public static CapacitySchedulerConfiguration setupQueueMappings( return conf; } + public static CapacitySchedulerConfiguration setupQueueMappingEntities( + CapacitySchedulerConfiguration conf, String queuePlacementRuleName, + String parentQueue, boolean overrideWithQueueMappings, int[] sourceIds) { + + List queuePlacementRules = new ArrayList<>(); + queuePlacementRules.add(queuePlacementRuleName); + conf.setQueuePlacementRules(queuePlacementRules); + + List existingMappings = + conf.getQueueMappingEntity(QUEUE_MAPPING_NAME); + + //set queue mapping + List queueMappings = + new ArrayList<>(); + for (int i = 0; i < sourceIds.length; i++) { + //Set C as parent queue name for auto queue creation + PlacementRule.QueueMappingEntity queueMapping = + new PlacementRule.QueueMappingEntity(USER + sourceIds[i], + getQueueMapping(parentQueue, USER + sourceIds[i])); + queueMappings.add(queueMapping); + } + + existingMappings.addAll(queueMappings); + conf.setQueueMappingEntities(existingMappings, QUEUE_MAPPING_NAME); + //override with queue mappings + conf.setOverrideWithQueueMappings(overrideWithQueueMappings); + return conf; + } + /** * @param conf, to be modified * @return, CS configuration which has C as an auto creation enabled parent @@ -682,4 +718,32 @@ private void validateQueueEntitlementChangesForLeafQueue(CSQueue leafQueue, } } + @Test + public void testUpdatePlacementRulesFactory() throws Exception { + CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration(); + setupQueueConfiguration(conf); + conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class, + ResourceScheduler.class); + + setupQueueMappingEntities(conf, QUEUE_MAPPING_RULE_APP_NAME, + PARENT_QUEUE, true, new int[] { 1, 2, 3}); + + mockRM = new MockRM(conf); + CapacityScheduler cs = (CapacityScheduler) mockRM.getResourceScheduler(); + cs.updatePlacementRules(); + mockRM.start(); + cs.start(); + + List rules = cs.getRMContext() + .getQueuePlacementManager().getPlacementRules(); + + List placementRuleNames = new ArrayList<>(); + for (PlacementRule pr : rules) { + placementRuleNames.add(pr.getName()); + } + + assertThat(placementRuleNames, hasItems(QUEUE_MAPPING_RULE_APP_NAME)); + + } + }