You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by rk...@apache.org on 2018/02/06 22:37:23 UTC
[2/2] hadoop git commit: YARN-7841. Cleanup
AllocationFileLoaderService's reloadAllocations method (snemeth via rkanter)
YARN-7841. Cleanup AllocationFileLoaderService's reloadAllocations method (snemeth via rkanter)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/814d701d
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/814d701d
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/814d701d
Branch: refs/heads/trunk
Commit: 814d701d46b4ff87f6ec94ba39667c80475c38d7
Parents: 4304fcd
Author: Robert Kanter <rk...@apache.org>
Authored: Tue Feb 6 14:36:49 2018 -0800
Committer: Robert Kanter <rk...@apache.org>
Committed: Tue Feb 6 14:36:49 2018 -0800
----------------------------------------------------------------------
.../scheduler/fair/AllocationConfiguration.java | 99 ++--
.../fair/AllocationFileLoaderService.java | 481 ++++---------------
.../fair/allocation/AllocationFileParser.java | 258 ++++++++++
.../allocation/AllocationFileQueueParser.java | 268 +++++++++++
.../fair/allocation/QueueProperties.java | 280 +++++++++++
.../fair/TestAllocationFileLoaderService.java | 187 ++++---
.../allocationfile/AllocationFileQueue.java | 82 ++++
.../AllocationFileQueueBuilder.java | 115 +++++
.../AllocationFileQueueProperties.java | 202 ++++++++
.../AllocationFileSimpleQueueBuilder.java | 64 +++
.../AllocationFileSubQueueBuilder.java | 54 +++
.../allocationfile/AllocationFileWriter.java | 175 +++++++
.../fair/allocationfile/UserSettings.java | 80 +++
13 files changed, 1798 insertions(+), 547 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/814d701d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationConfiguration.java
----------------------------------------------------------------------
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/fair/AllocationConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationConfiguration.java
index 3505bca..c98aadc 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationConfiguration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationConfiguration.java
@@ -31,6 +31,8 @@ import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.security.AccessType;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSchedulerConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.allocation.AllocationFileParser;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.allocation.QueueProperties;
import org.apache.hadoop.yarn.util.resource.Resources;
import com.google.common.annotations.VisibleForTesting;
@@ -47,7 +49,7 @@ public class AllocationConfiguration extends ReservationSchedulerConfiguration {
private final Map<String, ConfigurableResource> maxChildQueueResources;
// Sharing weights for each queue
private final Map<String, Float> queueWeights;
-
+
// Max concurrent running applications for each queue and for each user; in addition,
// for users that have no max specified, we use the userMaxJobsDefault.
@VisibleForTesting
@@ -88,13 +90,13 @@ public class AllocationConfiguration extends ReservationSchedulerConfiguration {
private final Set<String> reservableQueues;
private final Map<String, SchedulingPolicy> schedulingPolicies;
-
+
private final SchedulingPolicy defaultSchedulingPolicy;
-
+
// Policy for mapping apps to queues
@VisibleForTesting
QueuePlacementPolicy placementPolicy;
-
+
//Configured queues in the alloc xml
@VisibleForTesting
Map<FSQueueType, Set<String>> configuredQueues;
@@ -104,53 +106,42 @@ public class AllocationConfiguration extends ReservationSchedulerConfiguration {
private final Set<String> nonPreemptableQueues;
- public AllocationConfiguration(Map<String, Resource> minQueueResources,
- Map<String, ConfigurableResource> maxQueueResources,
- Map<String, ConfigurableResource> maxChildQueueResources,
- Map<String, Integer> queueMaxApps,
- Map<String, Integer> userMaxApps,
- Map<String, Float> queueWeights,
- Map<String, Float> queueMaxAMShares, int userMaxAppsDefault,
- int queueMaxAppsDefault,
- ConfigurableResource queueMaxResourcesDefault,
- float queueMaxAMShareDefault,
- Map<String, SchedulingPolicy> schedulingPolicies,
- SchedulingPolicy defaultSchedulingPolicy,
- Map<String, Long> minSharePreemptionTimeouts,
- Map<String, Long> fairSharePreemptionTimeouts,
- Map<String, Float> fairSharePreemptionThresholds,
- Map<String, Map<AccessType, AccessControlList>> queueAcls,
- Map<String, Map<ReservationACL, AccessControlList>> resAcls,
- QueuePlacementPolicy placementPolicy,
- Map<FSQueueType, Set<String>> configuredQueues,
- ReservationQueueConfiguration globalReservationQueueConfig,
- Set<String> reservableQueues,
- Set<String> nonPreemptableQueues) {
- this.minQueueResources = minQueueResources;
- this.maxQueueResources = maxQueueResources;
- this.maxChildQueueResources = maxChildQueueResources;
- this.queueMaxApps = queueMaxApps;
- this.userMaxApps = userMaxApps;
- this.queueMaxAMShares = queueMaxAMShares;
- this.queueWeights = queueWeights;
- this.userMaxAppsDefault = userMaxAppsDefault;
- this.queueMaxResourcesDefault = queueMaxResourcesDefault;
- this.queueMaxAppsDefault = queueMaxAppsDefault;
- this.queueMaxAMShareDefault = queueMaxAMShareDefault;
- this.defaultSchedulingPolicy = defaultSchedulingPolicy;
- this.schedulingPolicies = schedulingPolicies;
- this.minSharePreemptionTimeouts = minSharePreemptionTimeouts;
- this.fairSharePreemptionTimeouts = fairSharePreemptionTimeouts;
- this.fairSharePreemptionThresholds = fairSharePreemptionThresholds;
- this.queueAcls = queueAcls;
- this.resAcls = resAcls;
- this.reservableQueues = reservableQueues;
+ public AllocationConfiguration(QueueProperties queueProperties,
+ AllocationFileParser allocationFileParser,
+ QueuePlacementPolicy newPlacementPolicy,
+ ReservationQueueConfiguration globalReservationQueueConfig)
+ throws AllocationConfigurationException {
+ this.minQueueResources = queueProperties.getMinQueueResources();
+ this.maxQueueResources = queueProperties.getMaxQueueResources();
+ this.maxChildQueueResources = queueProperties.getMaxChildQueueResources();
+ this.queueMaxApps = queueProperties.getQueueMaxApps();
+ this.userMaxApps = allocationFileParser.getUserMaxApps();
+ this.queueMaxAMShares = queueProperties.getQueueMaxAMShares();
+ this.queueWeights = queueProperties.getQueueWeights();
+ this.userMaxAppsDefault = allocationFileParser.getUserMaxAppsDefault();
+ this.queueMaxResourcesDefault =
+ allocationFileParser.getQueueMaxResourcesDefault();
+ this.queueMaxAppsDefault = allocationFileParser.getQueueMaxAppsDefault();
+ this.queueMaxAMShareDefault =
+ allocationFileParser.getQueueMaxAMShareDefault();
+ this.defaultSchedulingPolicy =
+ allocationFileParser.getDefaultSchedulingPolicy();
+ this.schedulingPolicies = queueProperties.getQueuePolicies();
+ this.minSharePreemptionTimeouts =
+ queueProperties.getMinSharePreemptionTimeouts();
+ this.fairSharePreemptionTimeouts =
+ queueProperties.getFairSharePreemptionTimeouts();
+ this.fairSharePreemptionThresholds =
+ queueProperties.getFairSharePreemptionThresholds();
+ this.queueAcls = queueProperties.getQueueAcls();
+ this.resAcls = queueProperties.getReservationAcls();
+ this.reservableQueues = queueProperties.getReservableQueues();
this.globalReservationQueueConfig = globalReservationQueueConfig;
- this.placementPolicy = placementPolicy;
- this.configuredQueues = configuredQueues;
- this.nonPreemptableQueues = nonPreemptableQueues;
+ this.placementPolicy = newPlacementPolicy;
+ this.configuredQueues = queueProperties.getConfiguredQueues();
+ this.nonPreemptableQueues = queueProperties.getNonPreemptableQueues();
}
-
+
public AllocationConfiguration(Configuration conf) {
minQueueResources = new HashMap<>();
maxChildQueueResources = new HashMap<>();
@@ -179,7 +170,7 @@ public class AllocationConfiguration extends ReservationSchedulerConfiguration {
QueuePlacementPolicy.fromConfiguration(conf, configuredQueues);
nonPreemptableQueues = new HashSet<>();
}
-
+
/**
* Get the ACLs associated with this queue. If a given ACL is not explicitly
* configured, include the default value for that ACL. The default for the
@@ -300,7 +291,7 @@ public class AllocationConfiguration extends ReservationSchedulerConfiguration {
}
return maxQueueResource;
}
-
+
/**
* Get the maximum resource allocation for children of the given queue.
*
@@ -317,15 +308,15 @@ public class AllocationConfiguration extends ReservationSchedulerConfiguration {
SchedulingPolicy policy = schedulingPolicies.get(queueName);
return (policy == null) ? defaultSchedulingPolicy : policy;
}
-
+
public SchedulingPolicy getDefaultSchedulingPolicy() {
return defaultSchedulingPolicy;
}
-
+
public Map<FSQueueType, Set<String>> getConfiguredQueues() {
return configuredQueues;
}
-
+
public QueuePlacementPolicy getPlacementPolicy() {
return placementPolicy;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/814d701d/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
----------------------------------------------------------------------
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/fair/AllocationFileLoaderService.java b/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 f73e05f..d8d9051 100644
--- a/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
+++ b/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
@@ -29,24 +29,20 @@ import org.apache.hadoop.fs.UnsupportedFileSystemException;
import org.apache.hadoop.security.authorize.AccessControlList;
import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.yarn.api.records.QueueACL;
-import org.apache.hadoop.yarn.api.records.ReservationACL;
-import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.security.AccessType;
import org.apache.hadoop.yarn.security.Permission;
import org.apache.hadoop.yarn.security.PrivilegedEntity;
import org.apache.hadoop.yarn.security.PrivilegedEntity.EntityType;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FifoPolicy;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.allocation.AllocationFileParser;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.allocation.AllocationFileQueueParser;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.allocation.QueueProperties;
import org.apache.hadoop.yarn.util.Clock;
import org.apache.hadoop.yarn.util.SystemClock;
-import org.apache.hadoop.yarn.util.resource.Resources;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
-import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
-import org.w3c.dom.Text;
import org.xml.sax.SAXException;
-
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
@@ -54,18 +50,18 @@ import java.io.IOException;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
+import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.allocation.AllocationFileQueueParser.EVERYBODY_ACL;
+import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.allocation.AllocationFileQueueParser.ROOT;
@Public
@Unstable
public class AllocationFileLoaderService extends AbstractService {
-
+
public static final Log LOG = LogFactory.getLog(
AllocationFileLoaderService.class.getName());
-
+
/** Time to wait between checks of the allocation file */
public static final long ALLOC_RELOAD_INTERVAL_MS = 10 * 1000;
@@ -80,11 +76,6 @@ public class AllocationFileLoaderService extends AbstractService {
//Permitted allocation file filesystems (case insensitive)
private static final String SUPPORTED_FS_REGEX =
"(?i)(hdfs)|(file)|(s3a)|(viewfs)";
- private static final String ROOT = "root";
- private static final AccessControlList EVERYBODY_ACL =
- new AccessControlList("*");
- private static final AccessControlList NOBODY_ACL =
- new AccessControlList(" ");
private final Clock clock;
@@ -97,10 +88,10 @@ public class AllocationFileLoaderService extends AbstractService {
private FileSystem fs;
private Listener reloadListener;
-
+
@VisibleForTesting
long reloadIntervalMs = ALLOC_RELOAD_INTERVAL_MS;
-
+
private Thread reloadThread;
private volatile boolean running = true;
@@ -114,7 +105,7 @@ public class AllocationFileLoaderService extends AbstractService {
super(AllocationFileLoaderService.class.getName());
this.clock = clock;
}
-
+
@Override
public void serviceInit(Configuration conf) throws Exception {
this.allocFile = getAllocationFile(conf);
@@ -161,7 +152,7 @@ public class AllocationFileLoaderService extends AbstractService {
}
super.serviceInit(conf);
}
-
+
@Override
public void serviceStart() throws Exception {
if (reloadThread != null) {
@@ -169,7 +160,7 @@ public class AllocationFileLoaderService extends AbstractService {
}
super.serviceStart();
}
-
+
@Override
public void serviceStop() throws Exception {
running = false;
@@ -183,7 +174,7 @@ public class AllocationFileLoaderService extends AbstractService {
}
super.serviceStop();
}
-
+
/**
* Path to XML file containing allocations. If the
* path is relative, it is searched for in the
@@ -215,11 +206,11 @@ public class AllocationFileLoaderService extends AbstractService {
}
return allocPath;
}
-
+
public synchronized void setReloadListener(Listener reloadListener) {
this.reloadListener = reloadListener;
}
-
+
/**
* Updates the allocation list from the allocation config file. This file is
* expected to be in the XML format specified in the design doc.
@@ -229,417 +220,111 @@ public class AllocationFileLoaderService extends AbstractService {
* @throws ParserConfigurationException if XML parser is misconfigured.
* @throws SAXException if config file is malformed.
*/
- public synchronized void reloadAllocations() throws IOException,
- ParserConfigurationException, SAXException,
+ public synchronized void reloadAllocations()
+ throws IOException, ParserConfigurationException, SAXException,
AllocationConfigurationException {
if (allocFile == null) {
reloadListener.onReload(null);
return;
}
LOG.info("Loading allocation file " + allocFile);
- // Create some temporary hashmaps to hold the new allocs, and we only save
- // them in our fields if we have parsed the entire allocs file successfully.
- Map<String, Resource> minQueueResources = new HashMap<>();
- Map<String, ConfigurableResource> maxQueueResources = new HashMap<>();
- Map<String, ConfigurableResource> maxChildQueueResources = new HashMap<>();
- Map<String, Integer> queueMaxApps = new HashMap<>();
- Map<String, Integer> userMaxApps = new HashMap<>();
- Map<String, Float> queueMaxAMShares = new HashMap<>();
- Map<String, Float> queueWeights = new HashMap<>();
- Map<String, SchedulingPolicy> queuePolicies = new HashMap<>();
- Map<String, Long> minSharePreemptionTimeouts = new HashMap<>();
- Map<String, Long> fairSharePreemptionTimeouts = new HashMap<>();
- Map<String, Float> fairSharePreemptionThresholds = new HashMap<>();
- Map<String, Map<AccessType, AccessControlList>> queueAcls =
- new HashMap<>();
- Map<String, Map<ReservationACL, AccessControlList>> reservationAcls =
- new HashMap<>();
- Set<String> reservableQueues = new HashSet<>();
- Set<String> nonPreemptableQueues = new HashSet<>();
- int userMaxAppsDefault = Integer.MAX_VALUE;
- int queueMaxAppsDefault = Integer.MAX_VALUE;
- ConfigurableResource queueMaxResourcesDefault =
- new ConfigurableResource(Resources.unbounded());
- float queueMaxAMShareDefault = 0.5f;
- long defaultFairSharePreemptionTimeout = Long.MAX_VALUE;
- long defaultMinSharePreemptionTimeout = Long.MAX_VALUE;
- float defaultFairSharePreemptionThreshold = 0.5f;
- SchedulingPolicy defaultSchedPolicy = SchedulingPolicy.DEFAULT_POLICY;
-
- // Reservation global configuration knobs
- String planner = null;
- String reservationAgent = null;
- String reservationAdmissionPolicy = null;
-
- QueuePlacementPolicy newPlacementPolicy = null;
-
- // Remember all queue names so we can display them on web UI, etc.
- // configuredQueues is segregated based on whether it is a leaf queue
- // or a parent queue. This information is used for creating queues
- // and also for making queue placement decisions(QueuePlacementRule.java).
- Map<FSQueueType, Set<String>> configuredQueues = new HashMap<>();
-
- for (FSQueueType queueType : FSQueueType.values()) {
- configuredQueues.put(queueType, new HashSet<>());
- }
// Read and parse the allocations file.
DocumentBuilderFactory docBuilderFactory =
- DocumentBuilderFactory.newInstance();
+ DocumentBuilderFactory.newInstance();
docBuilderFactory.setIgnoringComments(true);
DocumentBuilder builder = docBuilderFactory.newDocumentBuilder();
Document doc = builder.parse(fs.open(allocFile));
Element root = doc.getDocumentElement();
- if (!"allocations".equals(root.getTagName()))
- throw new AllocationConfigurationException("Bad fair scheduler config " +
- "file: top-level element not <allocations>");
- NodeList elements = root.getChildNodes();
- List<Element> queueElements = new ArrayList<>();
- Element placementPolicyElement = null;
- for (int i = 0; i < elements.getLength(); i++) {
- Node node = elements.item(i);
- if (node instanceof Element) {
- Element element = (Element)node;
- if ("queue".equals(element.getTagName()) ||
- "pool".equals(element.getTagName())) {
- queueElements.add(element);
- } else if ("user".equals(element.getTagName())) {
- String userName = element.getAttribute("name");
- NodeList fields = element.getChildNodes();
- for (int j = 0; j < fields.getLength(); j++) {
- Node fieldNode = fields.item(j);
- if (!(fieldNode instanceof Element)) {
- continue;
- }
- Element field = (Element) fieldNode;
- if ("maxRunningApps".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData().trim();
- int val = Integer.parseInt(text);
- userMaxApps.put(userName, val);
- }
- }
- } else if ("queueMaxResourcesDefault".equals(element.getTagName())) {
- String text = ((Text)element.getFirstChild()).getData().trim();
- ConfigurableResource val =
- FairSchedulerConfiguration.parseResourceConfigValue(text);
- queueMaxResourcesDefault = val;
- } else if ("userMaxAppsDefault".equals(element.getTagName())) {
- String text = ((Text)element.getFirstChild()).getData().trim();
- int val = Integer.parseInt(text);
- userMaxAppsDefault = val;
- } else if ("defaultFairSharePreemptionTimeout"
- .equals(element.getTagName())) {
- String text = ((Text)element.getFirstChild()).getData().trim();
- long val = Long.parseLong(text) * 1000L;
- defaultFairSharePreemptionTimeout = val;
- } else if ("fairSharePreemptionTimeout".equals(element.getTagName())) {
- if (defaultFairSharePreemptionTimeout == Long.MAX_VALUE) {
- String text = ((Text)element.getFirstChild()).getData().trim();
- long val = Long.parseLong(text) * 1000L;
- defaultFairSharePreemptionTimeout = val;
- }
- } else if ("defaultMinSharePreemptionTimeout"
- .equals(element.getTagName())) {
- String text = ((Text)element.getFirstChild()).getData().trim();
- long val = Long.parseLong(text) * 1000L;
- defaultMinSharePreemptionTimeout = val;
- } else if ("defaultFairSharePreemptionThreshold"
- .equals(element.getTagName())) {
- String text = ((Text)element.getFirstChild()).getData().trim();
- float val = Float.parseFloat(text);
- val = Math.max(Math.min(val, 1.0f), 0.0f);
- defaultFairSharePreemptionThreshold = val;
- } else if ("queueMaxAppsDefault".equals(element.getTagName())) {
- String text = ((Text)element.getFirstChild()).getData().trim();
- int val = Integer.parseInt(text);
- queueMaxAppsDefault = val;
- } else if ("queueMaxAMShareDefault".equals(element.getTagName())) {
- String text = ((Text)element.getFirstChild()).getData().trim();
- float val = Float.parseFloat(text);
- val = Math.min(val, 1.0f);
- queueMaxAMShareDefault = val;
- } else if ("defaultQueueSchedulingPolicy".equals(element.getTagName())
- || "defaultQueueSchedulingMode".equals(element.getTagName())) {
- String text = ((Text)element.getFirstChild()).getData().trim();
- if (text.equalsIgnoreCase(FifoPolicy.NAME)) {
- throw new AllocationConfigurationException("Bad fair scheduler "
- + "config file: defaultQueueSchedulingPolicy or "
- + "defaultQueueSchedulingMode can't be FIFO.");
- }
- defaultSchedPolicy = SchedulingPolicy.parse(text);
- } else if ("queuePlacementPolicy".equals(element.getTagName())) {
- placementPolicyElement = element;
- } else if ("reservation-planner".equals(element.getTagName())) {
- String text = ((Text) element.getFirstChild()).getData().trim();
- planner = text;
- } else if ("reservation-agent".equals(element.getTagName())) {
- String text = ((Text) element.getFirstChild()).getData().trim();
- reservationAgent = text;
- } else if ("reservation-policy".equals(element.getTagName())) {
- String text = ((Text) element.getFirstChild()).getData().trim();
- reservationAdmissionPolicy = text;
- } else {
- LOG.warn("Bad element in allocations file: " + element.getTagName());
- }
- }
+ if (!"allocations".equals(root.getTagName())) {
+ throw new AllocationConfigurationException("Bad fair scheduler config "
+ + "file: top-level element not <allocations>");
}
+ NodeList elements = root.getChildNodes();
- // Load queue elements. A root queue can either be included or omitted. If
- // it's included, all other queues must be inside it.
- for (Element element : queueElements) {
- String parent = "root";
- if (element.getAttribute("name").equalsIgnoreCase("root")) {
- if (queueElements.size() > 1) {
- throw new AllocationConfigurationException("If configuring root queue,"
- + " no other queues can be placed alongside it.");
- }
- parent = null;
- }
- loadQueue(parent, element, minQueueResources, maxQueueResources,
- maxChildQueueResources, queueMaxApps, userMaxApps, queueMaxAMShares,
- queueWeights, queuePolicies, minSharePreemptionTimeouts,
- fairSharePreemptionTimeouts, fairSharePreemptionThresholds, queueAcls,
- reservationAcls, configuredQueues, reservableQueues,
- nonPreemptableQueues);
- }
+ AllocationFileParser allocationFileParser =
+ new AllocationFileParser(elements);
+ allocationFileParser.parse();
+
+ AllocationFileQueueParser queueParser =
+ new AllocationFileQueueParser(allocationFileParser.getQueueElements());
+ QueueProperties queueProperties = queueParser.parse();
// Load placement policy and pass it configured queues
Configuration conf = getConfig();
- if (placementPolicyElement != null) {
- newPlacementPolicy = QueuePlacementPolicy.fromXml(placementPolicyElement,
- configuredQueues, conf);
- } else {
- newPlacementPolicy = QueuePlacementPolicy.fromConfiguration(conf,
- configuredQueues);
- }
+ QueuePlacementPolicy newPlacementPolicy =
+ getQueuePlacementPolicy(allocationFileParser, queueProperties, conf);
+ setupRootQueueProperties(allocationFileParser, queueProperties);
- // Set the min/fair share preemption timeout for the root queue
- if (!minSharePreemptionTimeouts.containsKey(QueueManager.ROOT_QUEUE)){
- minSharePreemptionTimeouts.put(QueueManager.ROOT_QUEUE,
- defaultMinSharePreemptionTimeout);
- }
- if (!fairSharePreemptionTimeouts.containsKey(QueueManager.ROOT_QUEUE)) {
- fairSharePreemptionTimeouts.put(QueueManager.ROOT_QUEUE,
- defaultFairSharePreemptionTimeout);
- }
+ ReservationQueueConfiguration globalReservationQueueConfig =
+ createReservationQueueConfig(allocationFileParser);
- // Set the fair share preemption threshold for the root queue
- if (!fairSharePreemptionThresholds.containsKey(QueueManager.ROOT_QUEUE)) {
- fairSharePreemptionThresholds.put(QueueManager.ROOT_QUEUE,
- defaultFairSharePreemptionThreshold);
- }
-
- ReservationQueueConfiguration globalReservationQueueConfig = new
- ReservationQueueConfiguration();
- if (planner != null) {
- globalReservationQueueConfig.setPlanner(planner);
- }
- if (reservationAdmissionPolicy != null) {
- globalReservationQueueConfig.setReservationAdmissionPolicy
- (reservationAdmissionPolicy);
- }
- if (reservationAgent != null) {
- globalReservationQueueConfig.setReservationAgent(reservationAgent);
- }
-
- AllocationConfiguration info =
- new AllocationConfiguration(minQueueResources, maxQueueResources,
- maxChildQueueResources, queueMaxApps, userMaxApps, queueWeights,
- queueMaxAMShares, userMaxAppsDefault, queueMaxAppsDefault,
- queueMaxResourcesDefault, queueMaxAMShareDefault, queuePolicies,
- defaultSchedPolicy, minSharePreemptionTimeouts,
- fairSharePreemptionTimeouts, fairSharePreemptionThresholds, queueAcls,
- reservationAcls, newPlacementPolicy, configuredQueues,
- globalReservationQueueConfig, reservableQueues, nonPreemptableQueues);
+ AllocationConfiguration info = new AllocationConfiguration(queueProperties,
+ allocationFileParser, newPlacementPolicy, globalReservationQueueConfig);
lastSuccessfulReload = clock.getTime();
lastReloadAttemptFailed = false;
reloadListener.onReload(info);
}
-
- /**
- * Loads a queue from a queue element in the configuration file
- */
- private void loadQueue(String parentName, Element element,
- Map<String, Resource> minQueueResources,
- Map<String, ConfigurableResource> maxQueueResources,
- Map<String, ConfigurableResource> maxChildQueueResources,
- Map<String, Integer> queueMaxApps,
- Map<String, Integer> userMaxApps,
- Map<String, Float> queueMaxAMShares,
- Map<String, Float> queueWeights,
- Map<String, SchedulingPolicy> queuePolicies,
- Map<String, Long> minSharePreemptionTimeouts,
- Map<String, Long> fairSharePreemptionTimeouts,
- Map<String, Float> fairSharePreemptionThresholds,
- Map<String, Map<AccessType, AccessControlList>> queueAcls,
- Map<String, Map<ReservationACL, AccessControlList>> resAcls,
- Map<FSQueueType, Set<String>> configuredQueues,
- Set<String> reservableQueues,
- Set<String> nonPreemptableQueues)
- throws AllocationConfigurationException {
- String queueName = FairSchedulerUtilities.trimQueueName(
- element.getAttribute("name"));
- if (queueName.contains(".")) {
- throw new AllocationConfigurationException("Bad fair scheduler config "
- + "file: queue name (" + queueName + ") shouldn't contain period.");
+ private QueuePlacementPolicy getQueuePlacementPolicy(
+ AllocationFileParser allocationFileParser,
+ QueueProperties queueProperties, Configuration conf)
+ throws AllocationConfigurationException {
+ if (allocationFileParser.getQueuePlacementPolicy().isPresent()) {
+ return QueuePlacementPolicy.fromXml(
+ allocationFileParser.getQueuePlacementPolicy().get(),
+ queueProperties.getConfiguredQueues(), conf);
+ } else {
+ return QueuePlacementPolicy.fromConfiguration(conf,
+ queueProperties.getConfiguredQueues());
}
+ }
- if (queueName.isEmpty()) {
- throw new AllocationConfigurationException("Bad fair scheduler config "
- + "file: queue name shouldn't be empty or "
- + "consist only of whitespace.");
+ private void setupRootQueueProperties(
+ AllocationFileParser allocationFileParser,
+ QueueProperties queueProperties) {
+ // Set the min/fair share preemption timeout for the root queue
+ if (!queueProperties.getMinSharePreemptionTimeouts()
+ .containsKey(QueueManager.ROOT_QUEUE)) {
+ queueProperties.getMinSharePreemptionTimeouts().put(
+ QueueManager.ROOT_QUEUE,
+ allocationFileParser.getDefaultMinSharePreemptionTimeout());
}
-
- if (parentName != null) {
- queueName = parentName + "." + queueName;
+ if (!queueProperties.getFairSharePreemptionTimeouts()
+ .containsKey(QueueManager.ROOT_QUEUE)) {
+ queueProperties.getFairSharePreemptionTimeouts().put(
+ QueueManager.ROOT_QUEUE,
+ allocationFileParser.getDefaultFairSharePreemptionTimeout());
}
- Map<AccessType, AccessControlList> acls = new HashMap<>();
- Map<ReservationACL, AccessControlList> racls = new HashMap<>();
- NodeList fields = element.getChildNodes();
- boolean isLeaf = true;
- boolean isReservable = false;
-
- for (int j = 0; j < fields.getLength(); j++) {
- Node fieldNode = fields.item(j);
- if (!(fieldNode instanceof Element)) {
- continue;
- }
- Element field = (Element) fieldNode;
- if ("minResources".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData().trim();
- ConfigurableResource val =
- FairSchedulerConfiguration.parseResourceConfigValue(text);
- minQueueResources.put(queueName, val.getResource());
- } else if ("maxResources".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData().trim();
- ConfigurableResource val =
- FairSchedulerConfiguration.parseResourceConfigValue(text);
- maxQueueResources.put(queueName, val);
- } else if ("maxChildResources".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData().trim();
- ConfigurableResource val =
- FairSchedulerConfiguration.parseResourceConfigValue(text);
- maxChildQueueResources.put(queueName, val);
- } else if ("maxRunningApps".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData().trim();
- int val = Integer.parseInt(text);
- queueMaxApps.put(queueName, val);
- } else if ("maxAMShare".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData().trim();
- float val = Float.parseFloat(text);
- val = Math.min(val, 1.0f);
- queueMaxAMShares.put(queueName, val);
- } else if ("weight".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData().trim();
- double val = Double.parseDouble(text);
- queueWeights.put(queueName, (float)val);
- } else if ("minSharePreemptionTimeout".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData().trim();
- long val = Long.parseLong(text) * 1000L;
- minSharePreemptionTimeouts.put(queueName, val);
- } else if ("fairSharePreemptionTimeout".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData().trim();
- long val = Long.parseLong(text) * 1000L;
- fairSharePreemptionTimeouts.put(queueName, val);
- } else if ("fairSharePreemptionThreshold".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData().trim();
- float val = Float.parseFloat(text);
- val = Math.max(Math.min(val, 1.0f), 0.0f);
- fairSharePreemptionThresholds.put(queueName, val);
- } else if ("schedulingPolicy".equals(field.getTagName())
- || "schedulingMode".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData().trim();
- SchedulingPolicy policy = SchedulingPolicy.parse(text);
- queuePolicies.put(queueName, policy);
- } else if ("aclSubmitApps".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData();
- acls.put(AccessType.SUBMIT_APP, new AccessControlList(text));
- } else if ("aclAdministerApps".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData();
- acls.put(AccessType.ADMINISTER_QUEUE, new AccessControlList(text));
- } else if ("aclAdministerReservations".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData();
- racls.put(ReservationACL.ADMINISTER_RESERVATIONS,
- new AccessControlList(text));
- } else if ("aclListReservations".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData();
- racls.put(ReservationACL.LIST_RESERVATIONS, new AccessControlList(
- text));
- } else if ("aclSubmitReservations".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData();
- racls.put(ReservationACL.SUBMIT_RESERVATIONS,
- new AccessControlList(text));
- } else if ("reservation".equals(field.getTagName())) {
- isReservable = true;
- reservableQueues.add(queueName);
- configuredQueues.get(FSQueueType.PARENT).add(queueName);
- } else if ("allowPreemptionFrom".equals(field.getTagName())) {
- String text = ((Text)field.getFirstChild()).getData().trim();
- if (!Boolean.parseBoolean(text)) {
- nonPreemptableQueues.add(queueName);
- }
- } else if ("queue".endsWith(field.getTagName()) ||
- "pool".equals(field.getTagName())) {
- loadQueue(queueName, field, minQueueResources, maxQueueResources,
- maxChildQueueResources, queueMaxApps, userMaxApps, queueMaxAMShares,
- queueWeights, queuePolicies, minSharePreemptionTimeouts,
- fairSharePreemptionTimeouts, fairSharePreemptionThresholds,
- queueAcls, resAcls, configuredQueues, reservableQueues,
- nonPreemptableQueues);
- isLeaf = false;
- }
+ // Set the fair share preemption threshold for the root queue
+ if (!queueProperties.getFairSharePreemptionThresholds()
+ .containsKey(QueueManager.ROOT_QUEUE)) {
+ queueProperties.getFairSharePreemptionThresholds().put(
+ QueueManager.ROOT_QUEUE,
+ allocationFileParser.getDefaultFairSharePreemptionThreshold());
}
+ }
- // if a leaf in the alloc file is marked as type='parent'
- // then store it as a parent queue
- if (isLeaf && !"parent".equals(element.getAttribute("type"))) {
- configuredQueues.get(FSQueueType.LEAF).add(queueName);
- } else {
- if (isReservable) {
- throw new AllocationConfigurationException("The configuration settings"
- + " for " + queueName + " are invalid. A queue element that "
- + "contains child queue elements or that has the type='parent' "
- + "attribute cannot also include a reservation element.");
- }
- configuredQueues.get(FSQueueType.PARENT).add(queueName);
+ private ReservationQueueConfiguration createReservationQueueConfig(
+ AllocationFileParser allocationFileParser) {
+ ReservationQueueConfiguration globalReservationQueueConfig =
+ new ReservationQueueConfiguration();
+ if (allocationFileParser.getReservationPlanner().isPresent()) {
+ globalReservationQueueConfig
+ .setPlanner(allocationFileParser.getReservationPlanner().get());
}
-
- // Set default acls if not defined
- // The root queue defaults to all access
- for (QueueACL acl : QueueACL.values()) {
- AccessType accessType = SchedulerUtils.toAccessType(acl);
- if (acls.get(accessType) == null){
- AccessControlList defaultAcl = queueName.equals(ROOT) ?
- EVERYBODY_ACL : NOBODY_ACL;
- acls.put(accessType, defaultAcl);
- }
+ if (allocationFileParser.getReservationAdmissionPolicy().isPresent()) {
+ globalReservationQueueConfig.setReservationAdmissionPolicy(
+ allocationFileParser.getReservationAdmissionPolicy().get());
}
-
- queueAcls.put(queueName, acls);
- resAcls.put(queueName, racls);
- checkMinAndMaxResource(minQueueResources, maxQueueResources, queueName);
- }
-
- private void checkMinAndMaxResource(Map<String, Resource> minResources,
- Map<String, ConfigurableResource> maxResources, String queueName) {
-
- ConfigurableResource maxConfigurableResource = maxResources.get(queueName);
- Resource minResource = minResources.get(queueName);
-
- if (maxConfigurableResource != null && minResource != null) {
- Resource maxResource = maxConfigurableResource.getResource();
-
- // check whether max resource is bigger or equals to min resource when max
- // resource are absolute values
- if (maxResource != null && !Resources.fitsIn(minResource, maxResource)) {
- LOG.warn(String.format("Queue %s has max resources %s less than "
- + "min resources %s", queueName, maxResource, minResource));
- }
+ if (allocationFileParser.getReservationAgent().isPresent()) {
+ globalReservationQueueConfig.setReservationAgent(
+ allocationFileParser.getReservationAgent().get());
}
+ return globalReservationQueueConfig;
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/814d701d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/allocation/AllocationFileParser.java
----------------------------------------------------------------------
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/fair/allocation/AllocationFileParser.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/allocation/AllocationFileParser.java
new file mode 100644
index 0000000..161405b
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/allocation/AllocationFileParser.java
@@ -0,0 +1,258 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.allocation;
+
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.AllocationConfigurationException;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.ConfigurableResource;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairSchedulerConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.SchedulingPolicy;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FifoPolicy;
+import org.apache.hadoop.yarn.util.resource.Resources;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.w3c.dom.Element;
+import org.w3c.dom.Node;
+import org.w3c.dom.NodeList;
+import org.w3c.dom.Text;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Responsible for parsing allocation.xml config file.
+ * All node's text value is stored to textValues if {@link #VALID_TAG_NAMES}
+ * contains the tag name.
+ * Other meaningful fields are also saved in {@link #parse()}.
+ */
+public class AllocationFileParser {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AllocationFileParser.class);
+
+ private static final String QUEUE_MAX_RESOURCES_DEFAULT =
+ "queueMaxResourcesDefault";
+ private static final String USER_MAX_APPS_DEFAULT = "userMaxAppsDefault";
+ private static final String DEFAULT_FAIR_SHARE_PREEMPTION_TIMEOUT =
+ "defaultFairSharePreemptionTimeout";
+ private static final String FAIR_SHARE_PREEMPTION_TIMEOUT =
+ "fairSharePreemptionTimeout";
+ private static final String DEFAULT_MIN_SHARE_PREEMPTION_TIMEOUT =
+ "defaultMinSharePreemptionTimeout";
+ private static final String QUEUE_MAX_APPS_DEFAULT = "queueMaxAppsDefault";
+ private static final String DEFAULT_FAIR_SHARE_PREEMPTION_THRESHOLD =
+ "defaultFairSharePreemptionThreshold";
+ private static final String QUEUE_MAX_AM_SHARE_DEFAULT =
+ "queueMaxAMShareDefault";
+ private static final String RESERVATION_PLANNER = "reservation-planner";
+ private static final String RESERVATION_AGENT = "reservation-agent";
+ private static final String RESERVATION_ADMISSION_POLICY =
+ "reservation-policy";
+ private static final String QUEUE_PLACEMENT_POLICY = "queuePlacementPolicy";
+ private static final String QUEUE = "queue";
+ private static final String POOL = "pool";
+ private static final String USER = "user";
+ private static final String USERNAME = "name";
+ private static final String MAX_RUNNING_APPS = "maxRunningApps";
+ private static final String DEFAULT_QUEUE_SCHEDULING_POLICY =
+ "defaultQueueSchedulingPolicy";
+ private static final String DEFAULT_QUEUE_SCHEDULING_MODE =
+ "defaultQueueSchedulingMode";
+
+ private static final Set<String> VALID_TAG_NAMES =
+ Sets.newHashSet(QUEUE_MAX_RESOURCES_DEFAULT, USER_MAX_APPS_DEFAULT,
+ DEFAULT_FAIR_SHARE_PREEMPTION_TIMEOUT, FAIR_SHARE_PREEMPTION_TIMEOUT,
+ DEFAULT_MIN_SHARE_PREEMPTION_TIMEOUT, QUEUE_MAX_APPS_DEFAULT,
+ DEFAULT_FAIR_SHARE_PREEMPTION_THRESHOLD, QUEUE_MAX_AM_SHARE_DEFAULT,
+ RESERVATION_PLANNER, RESERVATION_AGENT, RESERVATION_ADMISSION_POLICY,
+ QUEUE_PLACEMENT_POLICY, QUEUE, POOL, USER,
+ DEFAULT_QUEUE_SCHEDULING_POLICY, DEFAULT_QUEUE_SCHEDULING_MODE);
+
+ private final NodeList elements;
+ private final Map<String, String> textValues = Maps.newHashMap();
+ private Element queuePlacementPolicyElement;
+ private final List<Element> queueElements = new ArrayList<>();
+ private final Map<String, Integer> userMaxApps = new HashMap<>();
+ private SchedulingPolicy defaultSchedulingPolicy;
+
+ public AllocationFileParser(NodeList elements) {
+ this.elements = elements;
+ }
+
+ public void parse() throws AllocationConfigurationException {
+ for (int i = 0; i < elements.getLength(); i++) {
+ Node node = elements.item(i);
+ if (node instanceof Element) {
+ Element element = (Element) node;
+ final String tagName = element.getTagName();
+ if (VALID_TAG_NAMES.contains(tagName)) {
+ if (tagName.equals(QUEUE_PLACEMENT_POLICY)) {
+ queuePlacementPolicyElement = element;
+ } else if (isSchedulingPolicy(element)) {
+ defaultSchedulingPolicy = extractSchedulingPolicy(element);
+ } else if (isQueue(element)) {
+ queueElements.add(element);
+ } else if (tagName.equals(USER)) {
+ extractUserData(element);
+ } else {
+ textValues.put(tagName, getTrimmedTextData(element));
+ }
+ } else {
+ LOG.warn("Bad element in allocations file: " + tagName);
+ }
+ }
+ }
+ }
+
+ private boolean isSchedulingPolicy(Element element) {
+ return DEFAULT_QUEUE_SCHEDULING_POLICY.equals(element.getTagName())
+ || DEFAULT_QUEUE_SCHEDULING_MODE.equals(element.getTagName());
+ }
+
+ private void extractUserData(Element element) {
+ final String userName = element.getAttribute(USERNAME);
+ final NodeList fields = element.getChildNodes();
+ for (int j = 0; j < fields.getLength(); j++) {
+ final Node fieldNode = fields.item(j);
+ if (!(fieldNode instanceof Element)) {
+ continue;
+ }
+ final Element field = (Element) fieldNode;
+ if (MAX_RUNNING_APPS.equals(field.getTagName())) {
+ final String text = getTrimmedTextData(field);
+ final int val = Integer.parseInt(text);
+ userMaxApps.put(userName, val);
+ }
+ }
+ }
+
+ private SchedulingPolicy extractSchedulingPolicy(Element element)
+ throws AllocationConfigurationException {
+ String text = getTrimmedTextData(element);
+ if (text.equalsIgnoreCase(FifoPolicy.NAME)) {
+ throw new AllocationConfigurationException("Bad fair scheduler "
+ + "config file: defaultQueueSchedulingPolicy or "
+ + "defaultQueueSchedulingMode can't be FIFO.");
+ }
+ return SchedulingPolicy.parse(text);
+ }
+
+ private boolean isQueue(Element element) {
+ return element.getTagName().equals(QUEUE)
+ || element.getTagName().equals(POOL);
+ }
+
+ private String getTrimmedTextData(Element element) {
+ return ((Text) element.getFirstChild()).getData().trim();
+ }
+
+ public ConfigurableResource getQueueMaxResourcesDefault()
+ throws AllocationConfigurationException {
+ Optional<String> value = getTextValue(QUEUE_MAX_RESOURCES_DEFAULT);
+ if (value.isPresent()) {
+ return FairSchedulerConfiguration.parseResourceConfigValue(value.get());
+ }
+ return new ConfigurableResource(Resources.unbounded());
+ }
+
+ public int getUserMaxAppsDefault() {
+ Optional<String> value = getTextValue(USER_MAX_APPS_DEFAULT);
+ return value.map(Integer::parseInt).orElse(Integer.MAX_VALUE);
+ }
+
+ public long getDefaultFairSharePreemptionTimeout() {
+ Optional<String> value = getTextValue(FAIR_SHARE_PREEMPTION_TIMEOUT);
+ Optional<String> defaultValue =
+ getTextValue(DEFAULT_FAIR_SHARE_PREEMPTION_TIMEOUT);
+
+ if (value.isPresent() && !defaultValue.isPresent()) {
+ return Long.parseLong(value.get()) * 1000L;
+ } else if (defaultValue.isPresent()) {
+ return Long.parseLong(defaultValue.get()) * 1000L;
+ }
+ return Long.MAX_VALUE;
+ }
+
+ public long getDefaultMinSharePreemptionTimeout() {
+ Optional<String> value = getTextValue(DEFAULT_MIN_SHARE_PREEMPTION_TIMEOUT);
+ return value.map(v -> Long.parseLong(v) * 1000L).orElse(Long.MAX_VALUE);
+ }
+
+ public int getQueueMaxAppsDefault() {
+ Optional<String> value = getTextValue(QUEUE_MAX_APPS_DEFAULT);
+ return value.map(Integer::parseInt).orElse(Integer.MAX_VALUE);
+ }
+
+ public float getDefaultFairSharePreemptionThreshold() {
+ Optional<String> value =
+ getTextValue(DEFAULT_FAIR_SHARE_PREEMPTION_THRESHOLD);
+ if (value.isPresent()) {
+ float floatValue = Float.parseFloat(value.get());
+ return Math.max(Math.min(floatValue, 1.0f), 0.0f);
+ }
+ return 0.5f;
+ }
+
+ public float getQueueMaxAMShareDefault() {
+ Optional<String> value = getTextValue(QUEUE_MAX_AM_SHARE_DEFAULT);
+ if (value.isPresent()) {
+ float val = Float.parseFloat(value.get());
+ return Math.min(val, 1.0f);
+ }
+ return 0.5f;
+ }
+
+ // Reservation global configuration knobs
+ public Optional<String> getReservationPlanner() {
+ return getTextValue(RESERVATION_PLANNER);
+ }
+
+ public Optional<String> getReservationAgent() {
+ return getTextValue(RESERVATION_AGENT);
+ }
+
+ public Optional<String> getReservationAdmissionPolicy() {
+ return getTextValue(RESERVATION_ADMISSION_POLICY);
+ }
+
+ public Optional<Element> getQueuePlacementPolicy() {
+ return Optional.ofNullable(queuePlacementPolicyElement);
+ }
+
+ private Optional<String> getTextValue(String key) {
+ return Optional.ofNullable(textValues.get(key));
+ }
+
+ public List<Element> getQueueElements() {
+ return queueElements;
+ }
+
+ public Map<String, Integer> getUserMaxApps() {
+ return userMaxApps;
+ }
+
+ public SchedulingPolicy getDefaultSchedulingPolicy() {
+ if (defaultSchedulingPolicy != null) {
+ return defaultSchedulingPolicy;
+ }
+ return SchedulingPolicy.DEFAULT_POLICY;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/814d701d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/allocation/AllocationFileQueueParser.java
----------------------------------------------------------------------
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/fair/allocation/AllocationFileQueueParser.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/allocation/AllocationFileQueueParser.java
new file mode 100644
index 0000000..ec7e4a4
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/allocation/AllocationFileQueueParser.java
@@ -0,0 +1,268 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.allocation;
+
+import org.apache.hadoop.security.authorize.AccessControlList;
+import org.apache.hadoop.yarn.api.records.QueueACL;
+import org.apache.hadoop.yarn.api.records.ReservationACL;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.security.AccessType;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.*;
+import org.apache.hadoop.yarn.util.resource.Resources;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.w3c.dom.Element;
+import org.w3c.dom.Node;
+import org.w3c.dom.NodeList;
+import org.w3c.dom.Text;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Responsible for loading queue configuration properties
+ * from a list of {@link Element}s containing queues.
+ */
+public class AllocationFileQueueParser {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(AllocationFileQueueParser.class);
+
+ public static final String ROOT = "root";
+ public static final AccessControlList EVERYBODY_ACL =
+ new AccessControlList("*");
+ static final AccessControlList NOBODY_ACL = new AccessControlList(" ");
+ private static final String MIN_RESOURCES = "minResources";
+ private static final String MAX_RESOURCES = "maxResources";
+ private static final String MAX_CHILD_RESOURCES = "maxChildResources";
+ private static final String MAX_RUNNING_APPS = "maxRunningApps";
+ private static final String MAX_AMSHARE = "maxAMShare";
+ private static final String WEIGHT = "weight";
+ private static final String MIN_SHARE_PREEMPTION_TIMEOUT =
+ "minSharePreemptionTimeout";
+ private static final String FAIR_SHARE_PREEMPTION_TIMEOUT =
+ "fairSharePreemptionTimeout";
+ private static final String FAIR_SHARE_PREEMPTION_THRESHOLD =
+ "fairSharePreemptionThreshold";
+ private static final String SCHEDULING_POLICY = "schedulingPolicy";
+ private static final String SCHEDULING_MODE = "schedulingMode";
+ private static final String ACL_SUBMIT_APPS = "aclSubmitApps";
+ private static final String ACL_ADMINISTER_APPS = "aclAdministerApps";
+ private static final String ACL_ADMINISTER_RESERVATIONS =
+ "aclAdministerReservations";
+ private static final String ACL_LIST_RESERVATIONS = "aclListReservations";
+ private static final String ACL_SUBMIT_RESERVATIONS = "aclSubmitReservations";
+ private static final String RESERVATION = "reservation";
+ private static final String ALLOW_PREEMPTION_FROM = "allowPreemptionFrom";
+ private static final String QUEUE = "queue";
+ private static final String POOL = "pool";
+
+ private final List<Element> elements;
+
+ public AllocationFileQueueParser(List<Element> elements) {
+ this.elements = elements;
+ }
+
+ // Load queue elements. A root queue can either be included or omitted. If
+ // it's included, all other queues must be inside it.
+ public QueueProperties parse() throws AllocationConfigurationException {
+ QueueProperties.Builder queuePropertiesBuilder =
+ new QueueProperties.Builder();
+ for (Element element : elements) {
+ String parent = ROOT;
+ if (element.getAttribute("name").equalsIgnoreCase(ROOT)) {
+ if (elements.size() > 1) {
+ throw new AllocationConfigurationException(
+ "If configuring root queue,"
+ + " no other queues can be placed alongside it.");
+ }
+ parent = null;
+ }
+ loadQueue(parent, element, queuePropertiesBuilder);
+ }
+
+ return queuePropertiesBuilder.build();
+ }
+
+ /**
+ * Loads a queue from a queue element in the configuration file.
+ */
+ private void loadQueue(String parentName, Element element,
+ QueueProperties.Builder builder) throws AllocationConfigurationException {
+ String queueName =
+ FairSchedulerUtilities.trimQueueName(element.getAttribute("name"));
+
+ if (queueName.contains(".")) {
+ throw new AllocationConfigurationException("Bad fair scheduler config "
+ + "file: queue name (" + queueName + ") shouldn't contain period.");
+ }
+
+ if (queueName.isEmpty()) {
+ throw new AllocationConfigurationException("Bad fair scheduler config "
+ + "file: queue name shouldn't be empty or "
+ + "consist only of whitespace.");
+ }
+
+ if (parentName != null) {
+ queueName = parentName + "." + queueName;
+ }
+
+ NodeList fields = element.getChildNodes();
+ boolean isLeaf = true;
+ boolean isReservable = false;
+
+ for (int j = 0; j < fields.getLength(); j++) {
+ Node fieldNode = fields.item(j);
+ if (!(fieldNode instanceof Element)) {
+ continue;
+ }
+ Element field = (Element) fieldNode;
+ if (MIN_RESOURCES.equals(field.getTagName())) {
+ String text = getTrimmedTextData(field);
+ ConfigurableResource val =
+ FairSchedulerConfiguration.parseResourceConfigValue(text);
+ builder.minQueueResources(queueName, val.getResource());
+ } else if (MAX_RESOURCES.equals(field.getTagName())) {
+ String text = getTrimmedTextData(field);
+ ConfigurableResource val =
+ FairSchedulerConfiguration.parseResourceConfigValue(text);
+ builder.maxQueueResources(queueName, val);
+ } else if (MAX_CHILD_RESOURCES.equals(field.getTagName())) {
+ String text = getTrimmedTextData(field);
+ ConfigurableResource val =
+ FairSchedulerConfiguration.parseResourceConfigValue(text);
+ builder.maxChildQueueResources(queueName, val);
+ } else if (MAX_RUNNING_APPS.equals(field.getTagName())) {
+ String text = getTrimmedTextData(field);
+ int val = Integer.parseInt(text);
+ builder.queueMaxApps(queueName, val);
+ } else if (MAX_AMSHARE.equals(field.getTagName())) {
+ String text = getTrimmedTextData(field);
+ float val = Float.parseFloat(text);
+ val = Math.min(val, 1.0f);
+ builder.queueMaxAMShares(queueName, val);
+ } else if (WEIGHT.equals(field.getTagName())) {
+ String text = getTrimmedTextData(field);
+ double val = Double.parseDouble(text);
+ builder.queueWeights(queueName, (float) val);
+ } else if (MIN_SHARE_PREEMPTION_TIMEOUT.equals(field.getTagName())) {
+ String text = getTrimmedTextData(field);
+ long val = Long.parseLong(text) * 1000L;
+ builder.minSharePreemptionTimeouts(queueName, val);
+ } else if (FAIR_SHARE_PREEMPTION_TIMEOUT.equals(field.getTagName())) {
+ String text = getTrimmedTextData(field);
+ long val = Long.parseLong(text) * 1000L;
+ builder.fairSharePreemptionTimeouts(queueName, val);
+ } else if (FAIR_SHARE_PREEMPTION_THRESHOLD.equals(field.getTagName())) {
+ String text = getTrimmedTextData(field);
+ float val = Float.parseFloat(text);
+ val = Math.max(Math.min(val, 1.0f), 0.0f);
+ builder.fairSharePreemptionThresholds(queueName, val);
+ } else if (SCHEDULING_POLICY.equals(field.getTagName())
+ || SCHEDULING_MODE.equals(field.getTagName())) {
+ String text = getTrimmedTextData(field);
+ SchedulingPolicy policy = SchedulingPolicy.parse(text);
+ builder.queuePolicies(queueName, policy);
+ } else if (ACL_SUBMIT_APPS.equals(field.getTagName())) {
+ String text = ((Text) field.getFirstChild()).getData();
+ builder.queueAcls(queueName, AccessType.SUBMIT_APP,
+ new AccessControlList(text));
+ } else if (ACL_ADMINISTER_APPS.equals(field.getTagName())) {
+ String text = ((Text) field.getFirstChild()).getData();
+ builder.queueAcls(queueName, AccessType.ADMINISTER_QUEUE,
+ new AccessControlList(text));
+ } else if (ACL_ADMINISTER_RESERVATIONS.equals(field.getTagName())) {
+ String text = ((Text) field.getFirstChild()).getData();
+ builder.reservationAcls(queueName,
+ ReservationACL.ADMINISTER_RESERVATIONS,
+ new AccessControlList(text));
+ } else if (ACL_LIST_RESERVATIONS.equals(field.getTagName())) {
+ String text = ((Text) field.getFirstChild()).getData();
+ builder.reservationAcls(queueName, ReservationACL.LIST_RESERVATIONS,
+ new AccessControlList(text));
+ } else if (ACL_SUBMIT_RESERVATIONS.equals(field.getTagName())) {
+ String text = ((Text) field.getFirstChild()).getData();
+ builder.reservationAcls(queueName, ReservationACL.SUBMIT_RESERVATIONS,
+ new AccessControlList(text));
+ } else if (RESERVATION.equals(field.getTagName())) {
+ isReservable = true;
+ builder.reservableQueues(queueName);
+ builder.configuredQueues(FSQueueType.PARENT, queueName);
+ } else if (ALLOW_PREEMPTION_FROM.equals(field.getTagName())) {
+ String text = getTrimmedTextData(field);
+ if (!Boolean.parseBoolean(text)) {
+ builder.nonPreemptableQueues(queueName);
+ }
+ } else if (QUEUE.endsWith(field.getTagName())
+ || POOL.equals(field.getTagName())) {
+ loadQueue(queueName, field, builder);
+ isLeaf = false;
+ }
+ }
+
+ // if a leaf in the alloc file is marked as type='parent'
+ // then store it as a parent queue
+ if (isLeaf && !"parent".equals(element.getAttribute("type"))) {
+ builder.configuredQueues(FSQueueType.LEAF, queueName);
+ } else {
+ if (isReservable) {
+ throw new AllocationConfigurationException("The configuration settings"
+ + " for " + queueName + " are invalid. A queue element that "
+ + "contains child queue elements or that has the type='parent' "
+ + "attribute cannot also include a reservation element.");
+ }
+ builder.configuredQueues(FSQueueType.PARENT, queueName);
+ }
+
+ // Set default acls if not defined
+ // The root queue defaults to all access
+ for (QueueACL acl : QueueACL.values()) {
+ AccessType accessType = SchedulerUtils.toAccessType(acl);
+ if (!builder.isAclDefinedForAccessType(queueName, accessType)) {
+ AccessControlList defaultAcl =
+ queueName.equals(ROOT) ? EVERYBODY_ACL : NOBODY_ACL;
+ builder.queueAcls(queueName, accessType, defaultAcl);
+ }
+ }
+
+ checkMinAndMaxResource(builder.getMinQueueResources(),
+ builder.getMaxQueueResources(), queueName);
+ }
+
+ private String getTrimmedTextData(Element element) {
+ return ((Text) element.getFirstChild()).getData().trim();
+ }
+
+ private void checkMinAndMaxResource(Map<String, Resource> minResources,
+ Map<String, ConfigurableResource> maxResources, String queueName) {
+
+ ConfigurableResource maxConfigurableResource = maxResources.get(queueName);
+ Resource minResource = minResources.get(queueName);
+
+ if (maxConfigurableResource != null && minResource != null) {
+ Resource maxResource = maxConfigurableResource.getResource();
+
+ // check whether max resource is greater or equals to min resource when
+ // max resource are absolute values
+ if (maxResource != null && !Resources.fitsIn(minResource, maxResource)) {
+ LOG.warn(String.format(
+ "Queue %s has max resources %s less than " + "min resources %s",
+ queueName, maxResource, minResource));
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/814d701d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/allocation/QueueProperties.java
----------------------------------------------------------------------
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/fair/allocation/QueueProperties.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/allocation/QueueProperties.java
new file mode 100644
index 0000000..ee5f179
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/allocation/QueueProperties.java
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.allocation;
+
+import org.apache.hadoop.security.authorize.AccessControlList;
+import org.apache.hadoop.yarn.api.records.ReservationACL;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.security.AccessType;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.ConfigurableResource;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSQueueType;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.SchedulingPolicy;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * This class is a value class, storing queue properties parsed from the
+ * allocation.xml config file. Since there are a bunch of properties, properties
+ * should be added via QueueProperties.Builder.
+ */
+public class QueueProperties {
+ // Create some temporary hashmaps to hold the new allocs, and we only save
+ // them in our fields if we have parsed the entire allocations file
+ // successfully.
+ private final Map<String, Resource> minQueueResources;
+ private final Map<String, ConfigurableResource> maxQueueResources;
+ private final Map<String, ConfigurableResource> maxChildQueueResources;
+ private final Map<String, Integer> queueMaxApps;
+ private final Map<String, Float> queueMaxAMShares;
+ private final Map<String, Float> queueWeights;
+ private final Map<String, SchedulingPolicy> queuePolicies;
+ private final Map<String, Long> minSharePreemptionTimeouts;
+ private final Map<String, Long> fairSharePreemptionTimeouts;
+ private final Map<String, Float> fairSharePreemptionThresholds;
+ private final Map<String, Map<AccessType, AccessControlList>> queueAcls;
+ private final Map<String, Map<ReservationACL, AccessControlList>>
+ reservationAcls;
+ private final Set<String> reservableQueues;
+ private final Set<String> nonPreemptableQueues;
+ private final Map<FSQueueType, Set<String>> configuredQueues;
+
+ QueueProperties(Builder builder) {
+ this.reservableQueues = builder.reservableQueues;
+ this.minQueueResources = builder.minQueueResources;
+ this.fairSharePreemptionTimeouts = builder.fairSharePreemptionTimeouts;
+ this.queueWeights = builder.queueWeights;
+ this.nonPreemptableQueues = builder.nonPreemptableQueues;
+ this.configuredQueues = builder.configuredQueues;
+ this.queueMaxAMShares = builder.queueMaxAMShares;
+ this.queuePolicies = builder.queuePolicies;
+ this.fairSharePreemptionThresholds = builder.fairSharePreemptionThresholds;
+ this.queueMaxApps = builder.queueMaxApps;
+ this.minSharePreemptionTimeouts = builder.minSharePreemptionTimeouts;
+ this.maxQueueResources = builder.maxQueueResources;
+ this.maxChildQueueResources = builder.maxChildQueueResources;
+ this.reservationAcls = builder.reservationAcls;
+ this.queueAcls = builder.queueAcls;
+ }
+
+ public Map<FSQueueType, Set<String>> getConfiguredQueues() {
+ return configuredQueues;
+ }
+
+ public Map<String, Long> getMinSharePreemptionTimeouts() {
+ return minSharePreemptionTimeouts;
+ }
+
+ public Map<String, Long> getFairSharePreemptionTimeouts() {
+ return fairSharePreemptionTimeouts;
+ }
+
+ public Map<String, Float> getFairSharePreemptionThresholds() {
+ return fairSharePreemptionThresholds;
+ }
+
+ public Map<String, Resource> getMinQueueResources() {
+ return minQueueResources;
+ }
+
+ public Map<String, ConfigurableResource> getMaxQueueResources() {
+ return maxQueueResources;
+ }
+
+ public Map<String, ConfigurableResource> getMaxChildQueueResources() {
+ return maxChildQueueResources;
+ }
+
+ public Map<String, Integer> getQueueMaxApps() {
+ return queueMaxApps;
+ }
+
+ public Map<String, Float> getQueueWeights() {
+ return queueWeights;
+ }
+
+ public Map<String, Float> getQueueMaxAMShares() {
+ return queueMaxAMShares;
+ }
+
+ public Map<String, SchedulingPolicy> getQueuePolicies() {
+ return queuePolicies;
+ }
+
+ public Map<String, Map<AccessType, AccessControlList>> getQueueAcls() {
+ return queueAcls;
+ }
+
+ public Map<String, Map<ReservationACL, AccessControlList>>
+ getReservationAcls() {
+ return reservationAcls;
+ }
+
+ public Set<String> getReservableQueues() {
+ return reservableQueues;
+ }
+
+ public Set<String> getNonPreemptableQueues() {
+ return nonPreemptableQueues;
+ }
+
+ /**
+ * Builder class for {@link QueueProperties}.
+ * All methods are adding queue properties to the maps of this builder
+ * keyed by the queue's name except some methods
+ * like {@link #isAclDefinedForAccessType(String, AccessType)} or
+ * {@link #getMinQueueResources()}.
+ *
+ */
+ public static final class Builder {
+ private Map<String, Resource> minQueueResources = new HashMap<>();
+ private Map<String, ConfigurableResource> maxQueueResources =
+ new HashMap<>();
+ private Map<String, ConfigurableResource> maxChildQueueResources =
+ new HashMap<>();
+ private Map<String, Integer> queueMaxApps = new HashMap<>();
+ private Map<String, Float> queueMaxAMShares = new HashMap<>();
+ private Map<String, Float> queueWeights = new HashMap<>();
+ private Map<String, SchedulingPolicy> queuePolicies = new HashMap<>();
+ private Map<String, Long> minSharePreemptionTimeouts = new HashMap<>();
+ private Map<String, Long> fairSharePreemptionTimeouts = new HashMap<>();
+ private Map<String, Float> fairSharePreemptionThresholds = new HashMap<>();
+ private Map<String, Map<AccessType, AccessControlList>> queueAcls =
+ new HashMap<>();
+ private Map<String, Map<ReservationACL, AccessControlList>>
+ reservationAcls = new HashMap<>();
+ private Set<String> reservableQueues = new HashSet<>();
+ private Set<String> nonPreemptableQueues = new HashSet<>();
+ // Remember all queue names so we can display them on web UI, etc.
+ // configuredQueues is segregated based on whether it is a leaf queue
+ // or a parent queue. This information is used for creating queues
+ // and also for making queue placement decisions(QueuePlacementRule.java).
+ private Map<FSQueueType, Set<String>> configuredQueues = new HashMap<>();
+
+ Builder() {
+ for (FSQueueType queueType : FSQueueType.values()) {
+ configuredQueues.put(queueType, new HashSet<>());
+ }
+ }
+
+ public static Builder create() {
+ return new Builder();
+ }
+
+ public Builder minQueueResources(String queueName, Resource resource) {
+ this.minQueueResources.put(queueName, resource);
+ return this;
+ }
+
+ public Builder maxQueueResources(String queueName,
+ ConfigurableResource resource) {
+ this.maxQueueResources.put(queueName, resource);
+ return this;
+ }
+
+ public Builder maxChildQueueResources(String queueName,
+ ConfigurableResource resource) {
+ this.maxChildQueueResources.put(queueName, resource);
+ return this;
+ }
+
+ public Builder queueMaxApps(String queueName, int value) {
+ this.queueMaxApps.put(queueName, value);
+ return this;
+ }
+
+ public Builder queueMaxAMShares(String queueName, float value) {
+ this.queueMaxAMShares.put(queueName, value);
+ return this;
+ }
+
+ public Builder queueWeights(String queueName, float value) {
+ this.queueWeights.put(queueName, value);
+ return this;
+ }
+
+ public Builder queuePolicies(String queueName, SchedulingPolicy policy) {
+ this.queuePolicies.put(queueName, policy);
+ return this;
+ }
+
+ public Builder minSharePreemptionTimeouts(String queueName, long value) {
+ this.minSharePreemptionTimeouts.put(queueName, value);
+ return this;
+ }
+
+ public Builder fairSharePreemptionTimeouts(String queueName, long value) {
+ this.fairSharePreemptionTimeouts.put(queueName, value);
+ return this;
+ }
+
+ public Builder fairSharePreemptionThresholds(String queueName,
+ float value) {
+ this.fairSharePreemptionThresholds.put(queueName, value);
+ return this;
+ }
+
+ public Builder queueAcls(String queueName, AccessType accessType,
+ AccessControlList acls) {
+ this.queueAcls.putIfAbsent(queueName, new HashMap<>());
+ this.queueAcls.get(queueName).put(accessType, acls);
+ return this;
+ }
+
+ public Builder reservationAcls(String queueName,
+ ReservationACL reservationACL, AccessControlList acls) {
+ this.reservationAcls.putIfAbsent(queueName, new HashMap<>());
+ this.reservationAcls.get(queueName).put(reservationACL, acls);
+ return this;
+ }
+
+ public Builder reservableQueues(String queue) {
+ this.reservableQueues.add(queue);
+ return this;
+ }
+
+ public Builder nonPreemptableQueues(String queue) {
+ this.nonPreemptableQueues.add(queue);
+ return this;
+ }
+
+ public void configuredQueues(FSQueueType queueType, String queueName) {
+ this.configuredQueues.get(queueType).add(queueName);
+ }
+
+ public boolean isAclDefinedForAccessType(String queueName,
+ AccessType accessType) {
+ Map<AccessType, AccessControlList> aclsForQueue =
+ this.queueAcls.get(queueName);
+ return aclsForQueue != null && aclsForQueue.get(accessType) != null;
+ }
+
+ public Map<String, Resource> getMinQueueResources() {
+ return minQueueResources;
+ }
+
+ public Map<String, ConfigurableResource> getMaxQueueResources() {
+ return maxQueueResources;
+ }
+
+ public QueueProperties build() {
+ return new QueueProperties(this);
+ }
+
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org