Repository: hadoop
Updated Branches:
refs/heads/trunk 7f6ed7fe3 -> c8212bacb
YARN-1707. Introduce APIs to add/remove/resize queues in the CapacityScheduler.
Contributed by Carlo Curino and Subru Krishnan
(cherry picked from commit aac47fda7fecda9fc18ade34d633eca895865a70)
Conflicts:
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
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/eb3e40b8
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/eb3e40b8
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/eb3e40b8
Branch: refs/heads/trunk
Commit: eb3e40b833b9d82c1556843f960194dc42e482f3
Parents: 7f6ed7f
Author: carlo curino <Carlo Curino>
Authored: Fri Sep 12 16:43:38 2014 -0700
Committer: Chris Douglas <[email protected]>
Committed: Fri Oct 3 15:41:02 2014 -0700
----------------------------------------------------------------------
.../scheduler/AbstractYarnScheduler.java | 25 ++
.../SchedulerDynamicEditException.java | 13 +
.../scheduler/YarnScheduler.java | 42 +++
.../scheduler/capacity/CapacityScheduler.java | 209 ++++++++++++--
.../scheduler/capacity/LeafQueue.java | 28 +-
.../scheduler/capacity/ParentQueue.java | 8 +-
.../scheduler/capacity/PlanQueue.java | 193 +++++++++++++
.../scheduler/capacity/ReservationQueue.java | 98 +++++++
.../scheduler/common/QueueEntitlement.java | 28 ++
.../webapp/dao/CapacitySchedulerQueueInfo.java | 9 +
.../capacity/TestCapacityScheduler.java | 2 +-
.../TestCapacitySchedulerDynamicBehavior.java | 282 +++++++++++++++++++
.../capacity/TestReservationQueue.java | 103 +++++++
13 files changed, 1012 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.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/AbstractYarnScheduler.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java
index 0b5447b..8e8d627 100644
---
a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java
+++
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java
@@ -51,6 +51,7 @@ import
org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppMoveEvent;
import
org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.QueueEntitlement;
import
org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
import
org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerFinishedEvent;
import
org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerImpl;
@@ -220,6 +221,24 @@ public abstract class AbstractYarnScheduler
+ " does not support moving apps between queues");
}
+ public void removeQueue(String queueName) throws YarnException {
+ throw new YarnException(getClass().getSimpleName()
+ + " does not support removing queues");
+ }
+
+ @Override
+ public void addQueue(Queue newQueue) throws YarnException {
+ throw new YarnException(getClass().getSimpleName()
+ + " does not support this operation");
+ }
+
+ @Override
+ public void setEntitlement(String queue, QueueEntitlement entitlement)
+ throws YarnException {
+ throw new YarnException(getClass().getSimpleName()
+ + " does not support this operation");
+ }
+
private void killOrphanContainerOnNode(RMNode node,
NMContainerStatus container) {
if (!container.getContainerState().equals(ContainerState.COMPLETE)) {
@@ -503,4 +522,10 @@ public abstract class AbstractYarnScheduler
public EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes() {
return EnumSet.of(SchedulerResourceTypes.MEMORY);
}
+
+ @Override
+ public Set<String> getPlanQueues() throws YarnException {
+ throw new YarnException(getClass().getSimpleName()
+ + " does not support reservations");
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerDynamicEditException.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/SchedulerDynamicEditException.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerDynamicEditException.java
new file mode 100644
index 0000000..42dc36a
--- /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/SchedulerDynamicEditException.java
@@ -0,0 +1,13 @@
+package org.apache.hadoop.yarn.server.resourcemanager.scheduler;
+
+import org.apache.hadoop.yarn.exceptions.YarnException;
+
+public class SchedulerDynamicEditException extends YarnException {
+
+ private static final long serialVersionUID = 7100374511387193257L;
+
+ public SchedulerDynamicEditException(String string) {
+ super(string);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/YarnScheduler.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/YarnScheduler.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/YarnScheduler.java
index b6c1018..d1b5275 100644
---
a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/YarnScheduler.java
+++
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/YarnScheduler.java
@@ -21,6 +21,7 @@ package
org.apache.hadoop.yarn.server.resourcemanager.scheduler;
import java.io.IOException;
import java.util.EnumSet;
import java.util.List;
+import java.util.Set;
import org.apache.hadoop.classification.InterfaceAudience.LimitedPrivate;
import org.apache.hadoop.classification.InterfaceAudience.Public;
@@ -41,6 +42,7 @@ import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.exceptions.YarnException;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.QueueEntitlement;
import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
import org.apache.hadoop.yarn.proto.YarnServiceProtos.SchedulerResourceTypes;
@@ -224,6 +226,46 @@ public interface YarnScheduler extends
EventHandler<SchedulerEvent> {
void killAllAppsInQueue(String queueName) throws YarnException;
/**
+ * Remove an existing queue. Implementations might limit when a queue could
be
+ * removed (e.g., must have zero entitlement, and no applications running, or
+ * must be a leaf, etc..).
+ *
+ * @param queueName name of the queue to remove
+ * @throws YarnException
+ */
+ void removeQueue(String queueName) throws YarnException;
+
+ /**
+ * Add to the scheduler a new Queue. Implementations might limit what type of
+ * queues can be dynamically added (e.g., Queue must be a leaf, must be
+ * attached to existing parent, must have zero entitlement).
+ *
+ * @param newQueue the queue being added.
+ * @throws YarnException
+ */
+ void addQueue(Queue newQueue) throws YarnException;
+
+ /**
+ * This method increase the entitlement for current queue (must respect
+ * invariants, e.g., no overcommit of parents, non negative, etc.).
+ * Entitlement is a general term for weights in FairScheduler, capacity for
+ * the CapacityScheduler, etc.
+ *
+ * @param queue the queue for which we change entitlement
+ * @param entitlement the new entitlement for the queue (capacity,
+ * maxCapacity, etc..)
+ * @throws YarnException
+ */
+ void setEntitlement(String queue, QueueEntitlement entitlement)
+ throws YarnException;
+
+ /**
+ * Gets the list of names for queues managed by the Reservation System
+ * @return the list of queues which support reservations
+ */
+ public Set<String> getPlanQueues() throws YarnException;
+
+ /**
* Return a collection of the resource types that are considered when
* scheduling
*
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/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
----------------------------------------------------------------------
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 e0816b5..ff6db3a 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
@@ -24,6 +24,8 @@ import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.HashSet;
+import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -90,6 +92,11 @@ import org.apache.hadoop.yarn.util.resource.Resources;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import org.apache.hadoop.yarn.api.records.ReservationId;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.QueueEntitlement;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerDynamicEditException;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Queue;
+
@LimitedPrivate("yarn")
@Evolving
@SuppressWarnings("unchecked")
@@ -473,9 +480,12 @@ public class CapacityScheduler extends
private void validateExistingQueues(
Map<String, CSQueue> queues, Map<String, CSQueue> newQueues)
throws IOException {
- for (String queue : queues.keySet()) {
- if (!newQueues.containsKey(queue)) {
- throw new IOException(queue + " cannot be found during refresh!");
+ // check that all static queues are included in the newQueues list
+ for (Map.Entry<String, CSQueue> e : queues.entrySet()) {
+ if (!(e.getValue() instanceof ReservationQueue)) {
+ if (!newQueues.containsKey(e.getKey())) {
+ throw new IOException(e.getKey() + " cannot be found during
refresh!");
+ }
}
}
}
@@ -507,26 +517,42 @@ public class CapacityScheduler extends
Map<String, CSQueue> oldQueues,
QueueHook hook) throws IOException {
CSQueue queue;
+ String fullQueueName =
+ (parent == null) ? queueName
+ : (parent.getQueuePath() + "." + queueName);
String[] childQueueNames =
- conf.getQueues((parent == null) ?
- queueName : (parent.getQueuePath()+"."+queueName));
+ conf.getQueues(fullQueueName);
+ boolean isReservableQueue = conf.isReservableQueue(fullQueueName);
if (childQueueNames == null || childQueueNames.length == 0) {
if (null == parent) {
throw new IllegalStateException(
"Queue configuration missing child queue names for " + queueName);
}
- queue =
- new LeafQueue(csContext, queueName, parent,
oldQueues.get(queueName));
-
- // Used only for unit tests
- queue = hook.hook(queue);
+ // Check if the queue will be dynamically managed by the Reservation
+ // system
+ if (isReservableQueue) {
+ queue =
+ new PlanQueue(csContext, queueName, parent,
+ oldQueues.get(queueName));
+ } else {
+ queue =
+ new LeafQueue(csContext, queueName, parent,
+ oldQueues.get(queueName));
+
+ // Used only for unit tests
+ queue = hook.hook(queue);
+ }
} else {
+ if (isReservableQueue) {
+ throw new IllegalStateException(
+ "Only Leaf Queues can be reservable for " + queueName);
+ }
ParentQueue parentQueue =
new ParentQueue(csContext, queueName, parent,
oldQueues.get(queueName));
// Used only for unit tests
queue = hook.hook(parentQueue);
-
+
List<CSQueue> childQueues = new ArrayList<CSQueue>();
for (String childQueueName : childQueueNames) {
CSQueue childQueue =
@@ -548,7 +574,7 @@ public class CapacityScheduler extends
return queue;
}
- synchronized CSQueue getQueue(String queueName) {
+ public synchronized CSQueue getQueue(String queueName) {
if (queueName == null) {
return null;
}
@@ -716,7 +742,7 @@ public class CapacityScheduler extends
ApplicationAttemptId applicationAttemptId,
RMAppAttemptState rmAppAttemptFinalState, boolean keepContainers) {
LOG.info("Application Attempt " + applicationAttemptId + " is done." +
- " finalState=" + rmAppAttemptFinalState);
+ " finalState=" + rmAppAttemptFinalState);
FiCaSchedulerApp attempt = getApplicationAttempt(applicationAttemptId);
SchedulerApplication<FiCaSchedulerApp> application =
@@ -996,9 +1022,16 @@ public class CapacityScheduler extends
case APP_ADDED:
{
AppAddedSchedulerEvent appAddedEvent = (AppAddedSchedulerEvent) event;
- addApplication(appAddedEvent.getApplicationId(),
- appAddedEvent.getQueue(),
- appAddedEvent.getUser(), appAddedEvent.getIsAppRecovering());
+ String queueName =
+ resolveReservationQueueName(appAddedEvent.getQueue(),
+ appAddedEvent.getApplicationId(),
+ appAddedEvent.getReservationID());
+ if (queueName != null) {
+ addApplication(appAddedEvent.getApplicationId(),
+ queueName,
+ appAddedEvent.getUser(),
+ appAddedEvent.getIsAppRecovering());
+ }
}
break;
case APP_REMOVED:
@@ -1231,6 +1264,123 @@ public class CapacityScheduler extends
}
}
+ private synchronized String resolveReservationQueueName(String queueName,
+ ApplicationId applicationId, ReservationId reservationID) {
+ CSQueue queue = getQueue(queueName);
+ // Check if the queue is a plan queue
+ if ((queue == null) || !(queue instanceof PlanQueue)) {
+ return queueName;
+ }
+ if (reservationID != null) {
+ String resQName = reservationID.toString();
+ queue = getQueue(resQName);
+ if (queue == null) {
+ String message =
+ "Application "
+ + applicationId
+ + " submitted to a reservation which is not yet currently
active: "
+ + resQName;
+ this.rmContext.getDispatcher().getEventHandler()
+ .handle(new RMAppRejectedEvent(applicationId, message));
+ return null;
+ }
+ // use the reservation queue to run the app
+ queueName = resQName;
+ } else {
+ // use the default child queue of the plan for unreserved apps
+ queueName = queueName + PlanQueue.DEFAULT_QUEUE_SUFFIX;
+ }
+ return queueName;
+ }
+
+ @Override
+ public synchronized void removeQueue(String queueName)
+ throws SchedulerDynamicEditException {
+ LOG.info("Removing queue: " + queueName);
+ CSQueue q = this.getQueue(queueName);
+ if (!(q instanceof ReservationQueue)) {
+ throw new SchedulerDynamicEditException("The queue that we are asked "
+ + "to remove (" + queueName + ") is not a ReservationQueue");
+ }
+ ReservationQueue disposableLeafQueue = (ReservationQueue) q;
+ // at this point we should have no more apps
+ if (disposableLeafQueue.getNumApplications() > 0) {
+ throw new SchedulerDynamicEditException("The queue " + queueName
+ + " is not empty " + disposableLeafQueue.getApplications().size()
+ + " active apps " + disposableLeafQueue.pendingApplications.size()
+ + " pending apps");
+ }
+
+ ((PlanQueue) disposableLeafQueue.getParent()).removeChildQueue(q);
+ this.queues.remove(queueName);
+ LOG.info("Removal of ReservationQueue " + queueName + " has succeeded");
+ }
+
+ @Override
+ public synchronized void addQueue(Queue queue)
+ throws SchedulerDynamicEditException {
+
+ if (!(queue instanceof ReservationQueue)) {
+ throw new SchedulerDynamicEditException("Queue " + queue.getQueueName()
+ + " is not a ReservationQueue");
+ }
+
+ ReservationQueue newQueue = (ReservationQueue) queue;
+
+ if (newQueue.getParent() == null
+ || !(newQueue.getParent() instanceof PlanQueue)) {
+ throw new SchedulerDynamicEditException("ParentQueue for "
+ + newQueue.getQueueName()
+ + " is not properly set (should be set and be a PlanQueue)");
+ }
+
+ PlanQueue parentPlan = (PlanQueue) newQueue.getParent();
+ String queuename = newQueue.getQueueName();
+ parentPlan.addChildQueue(newQueue);
+ this.queues.put(queuename, newQueue);
+ LOG.info("Creation of ReservationQueue " + newQueue + " succeeded");
+ }
+
+ @Override
+ public synchronized void setEntitlement(String inQueue,
+ QueueEntitlement entitlement) throws SchedulerDynamicEditException,
+ YarnException {
+ LeafQueue queue = getAndCheckLeafQueue(inQueue);
+ ParentQueue parent = (ParentQueue) queue.getParent();
+
+ if (!(queue instanceof ReservationQueue)) {
+ throw new SchedulerDynamicEditException("Entitlement can not be"
+ + " modified dynamically since queue " + inQueue
+ + " is not a ReservationQueue");
+ }
+
+ if (!(parent instanceof PlanQueue)) {
+ throw new SchedulerDynamicEditException("The parent of ReservationQueue "
+ + inQueue + " must be an PlanQueue");
+ }
+
+ ReservationQueue newQueue = (ReservationQueue) queue;
+
+ float sumChilds = ((PlanQueue) parent).sumOfChildCapacities();
+ float newChildCap = sumChilds - queue.getCapacity() +
entitlement.getCapacity();
+
+ if (newChildCap >= 0 && newChildCap < 1.0f + CSQueueUtils.EPSILON) {
+ // note: epsilon checks here are not ok, as the epsilons might accumulate
+ // and become a problem in aggregate
+ if (Math.abs(entitlement.getCapacity() - queue.getCapacity()) == 0
+ && Math.abs(entitlement.getMaxCapacity() -
queue.getMaximumCapacity()) == 0) {
+ return;
+ }
+ newQueue.setEntitlement(entitlement);
+ } else {
+ throw new SchedulerDynamicEditException(
+ "Sum of child queues would exceed 100% for PlanQueue: "
+ + parent.getQueueName());
+ }
+ LOG.info("Set entitlement for ReservationQueue " + inQueue + " to "
+ + queue.getCapacity() + " request was (" + entitlement.getCapacity() +
")");
+ }
+
@Override
public synchronized String moveApplication(ApplicationId appId,
String targetQueueName) throws YarnException {
@@ -1238,11 +1388,12 @@ public class CapacityScheduler extends
getApplicationAttempt(ApplicationAttemptId.newInstance(appId, 0));
String sourceQueueName = app.getQueue().getQueueName();
LeafQueue source = getAndCheckLeafQueue(sourceQueueName);
- LeafQueue dest = getAndCheckLeafQueue(targetQueueName);
+ String destQueueName = handleMoveToPlanQueue(targetQueueName);
+ LeafQueue dest = getAndCheckLeafQueue(destQueueName);
// Validation check - ACLs, submission limits for user & queue
String user = app.getUser();
try {
- dest.submitApplication(appId, user, targetQueueName);
+ dest.submitApplication(appId, user, destQueueName);
} catch (AccessControlException e) {
throw new YarnException(e);
}
@@ -1261,7 +1412,7 @@ public class CapacityScheduler extends
dest.submitApplicationAttempt(app, user);
applications.get(appId).setQueue(dest);
LOG.info("App: " + app.getApplicationId() + " successfully moved from "
- + sourceQueueName + " to: " + targetQueueName);
+ + sourceQueueName + " to: " + destQueueName);
return targetQueueName;
}
@@ -1296,4 +1447,24 @@ public class CapacityScheduler extends
return EnumSet
.of(SchedulerResourceTypes.MEMORY, SchedulerResourceTypes.CPU);
}
+
+ private String handleMoveToPlanQueue(String targetQueueName) {
+ CSQueue dest = getQueue(targetQueueName);
+ if (dest != null && dest instanceof PlanQueue) {
+ // use the default child reservation queue of the plan
+ targetQueueName = targetQueueName + PlanQueue.DEFAULT_QUEUE_SUFFIX;
+ }
+ return targetQueueName;
+ }
+
+ @Override
+ public Set<String> getPlanQueues() {
+ Set<String> ret = new HashSet<String>();
+ for (Map.Entry<String, CSQueue> l : queues.entrySet()) {
+ if (l.getValue() instanceof PlanQueue) {
+ ret.add(l.getKey());
+ }
+ }
+ return ret;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.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/capacity/LeafQueue.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/LeafQueue.java
index cdb6553..b9f5d5f 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/LeafQueue.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/LeafQueue.java
@@ -86,8 +86,8 @@ public class LeafQueue implements CSQueue {
private int userLimit;
private float userLimitFactor;
- private int maxApplications;
- private int maxApplicationsPerUser;
+ protected int maxApplications;
+ protected int maxApplicationsPerUser;
private float maxAMResourcePerQueuePercent;
private int maxActiveApplications; // Based on absolute max capacity
@@ -153,8 +153,7 @@ public class LeafQueue implements CSQueue {
Resources.subtract(maximumAllocation, minimumAllocation),
maximumAllocation);
- float capacity =
- (float)cs.getConfiguration().getCapacity(getQueuePath()) / 100;
+ float capacity = getCapacityFromConf();
float absoluteCapacity = parent.getAbsoluteCapacity() * capacity;
float maximumCapacity =
@@ -221,6 +220,11 @@ public class LeafQueue implements CSQueue {
this.activeApplications = new
TreeSet<FiCaSchedulerApp>(applicationComparator);
}
+ // externalizing in method, to allow overriding
+ protected float getCapacityFromConf() {
+ return (float)scheduler.getConfiguration().getCapacity(getQueuePath()) /
100;
+ }
+
private synchronized void setupQueueConfigs(
Resource clusterResource,
float capacity, float absoluteCapacity,
@@ -483,7 +487,7 @@ public class LeafQueue implements CSQueue {
* Set user limit factor - used only for testing.
* @param userLimitFactor new user limit factor
*/
- synchronized void setUserLimitFactor(int userLimitFactor) {
+ synchronized void setUserLimitFactor(float userLimitFactor) {
this.userLimitFactor = userLimitFactor;
}
@@ -831,7 +835,7 @@ public class LeafQueue implements CSQueue {
getApplication(reservedContainer.getApplicationAttemptId());
synchronized (application) {
return assignReservedContainer(application, node, reservedContainer,
- clusterResource);
+ clusterResource);
}
}
@@ -1880,4 +1884,16 @@ public class LeafQueue implements CSQueue {
getParent().detachContainer(clusterResource, application, rmContainer);
}
}
+
+ public void setCapacity(float capacity) {
+ this.capacity = capacity;
+ }
+
+ public void setAbsoluteCapacity(float absoluteCapacity) {
+ this.absoluteCapacity = absoluteCapacity;
+ }
+
+ public void setMaxApplications(int maxApplications) {
+ this.maxApplications = maxApplications;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.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/capacity/ParentQueue.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/ParentQueue.java
index aa74be1..011c99c 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/ParentQueue.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/ParentQueue.java
@@ -75,7 +75,7 @@ public class ParentQueue implements CSQueue {
private float usedCapacity = 0.0f;
- private final Set<CSQueue> childQueues;
+ protected final Set<CSQueue> childQueues;
private final Comparator<CSQueue> queueComparator;
private Resource usedResources = Resources.createResource(0, 0);
@@ -159,7 +159,7 @@ public class ParentQueue implements CSQueue {
", fullname=" + getQueuePath());
}
- private synchronized void setupQueueConfigs(
+ protected synchronized void setupQueueConfigs(
Resource clusterResource,
float capacity, float absoluteCapacity,
float maximumCapacity, float absoluteMaxCapacity,
@@ -881,4 +881,8 @@ public class ParentQueue implements CSQueue {
}
}
}
+
+ public Map<QueueACL, AccessControlList> getACLs() {
+ return acls;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/PlanQueue.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/capacity/PlanQueue.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/PlanQueue.java
new file mode 100644
index 0000000..4ada778
--- /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/capacity/PlanQueue.java
@@ -0,0 +1,193 @@
+package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.hadoop.yarn.api.records.Resource;
+import
org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSystem;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerDynamicEditException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This represents a dynamic queue managed by the {@link ReservationSystem}.
+ * From the user perspective this is equivalent to a LeafQueue that respect
+ * reservations, but functionality wise is a sub-class of ParentQueue
+ *
+ */
+public class PlanQueue extends ParentQueue {
+
+ public static final String DEFAULT_QUEUE_SUFFIX = "-default";
+
+ private static final Logger LOG = LoggerFactory.getLogger(PlanQueue.class);
+
+ private int maxAppsForReservation;
+ private int maxAppsPerUserForReservation;
+ private int userLimit;
+ private float userLimitFactor;
+ protected CapacitySchedulerContext schedulerContext;
+ private boolean showReservationsAsQueues;
+
+ public PlanQueue(CapacitySchedulerContext cs, String queueName,
+ CSQueue parent, CSQueue old) {
+ super(cs, queueName, parent, old);
+
+ this.schedulerContext = cs;
+ // Set the reservation queue attributes for the Plan
+ CapacitySchedulerConfiguration conf = cs.getConfiguration();
+ String queuePath = super.getQueuePath();
+ int maxAppsForReservation = conf.getMaximumApplicationsPerQueue(queuePath);
+ showReservationsAsQueues = conf.getShowReservationAsQueues(queuePath);
+ if (maxAppsForReservation < 0) {
+ maxAppsForReservation =
+ (int)
(CapacitySchedulerConfiguration.DEFAULT_MAXIMUM_SYSTEM_APPLICATIIONS * super
+ .getAbsoluteCapacity());
+ }
+ int userLimit = conf.getUserLimit(queuePath);
+ float userLimitFactor = conf.getUserLimitFactor(queuePath);
+ int maxAppsPerUserForReservation =
+ (int) (maxAppsForReservation * (userLimit / 100.0f) * userLimitFactor);
+ updateQuotas(userLimit, userLimitFactor, maxAppsForReservation,
+ maxAppsPerUserForReservation);
+
+ StringBuffer queueInfo = new StringBuffer();
+ queueInfo.append("Created Plan Queue: ").append(queueName)
+ .append("\nwith capacity: [").append(super.getCapacity())
+ .append("]\nwith max capacity: [").append(super.getMaximumCapacity())
+ .append("\nwith max reservation apps: [").append(maxAppsForReservation)
+ .append("]\nwith max reservation apps per user: [")
+ .append(maxAppsPerUserForReservation).append("]\nwith user limit: [")
+ .append(userLimit).append("]\nwith user limit factor: [")
+ .append(userLimitFactor).append("].");
+ LOG.info(queueInfo.toString());
+ }
+
+ @Override
+ public synchronized void reinitialize(CSQueue newlyParsedQueue,
+ Resource clusterResource) throws IOException {
+ // Sanity check
+ if (!(newlyParsedQueue instanceof PlanQueue)
+ || !newlyParsedQueue.getQueuePath().equals(getQueuePath())) {
+ throw new IOException("Trying to reinitialize " + getQueuePath()
+ + " from " + newlyParsedQueue.getQueuePath());
+ }
+
+ PlanQueue newlyParsedParentQueue = (PlanQueue) newlyParsedQueue;
+
+ if (newlyParsedParentQueue.getChildQueues().size() > 0) {
+ throw new IOException(
+ "Reservable Queue should not have sub-queues in the"
+ + "configuration");
+ }
+
+ // Set new configs
+ setupQueueConfigs(clusterResource, newlyParsedParentQueue.getCapacity(),
+ newlyParsedParentQueue.getAbsoluteCapacity(),
+ newlyParsedParentQueue.getMaximumCapacity(),
+ newlyParsedParentQueue.getAbsoluteMaximumCapacity(),
+ newlyParsedParentQueue.getState(), newlyParsedParentQueue.getACLs());
+
+ updateQuotas(newlyParsedParentQueue.userLimit,
+ newlyParsedParentQueue.userLimitFactor,
+ newlyParsedParentQueue.maxAppsForReservation,
+ newlyParsedParentQueue.maxAppsPerUserForReservation);
+
+ // run reinitialize on each existing queue, to trigger absolute cap
+ // recomputations
+ for (CSQueue res : this.getChildQueues()) {
+ res.reinitialize(res, clusterResource);
+ }
+ showReservationsAsQueues = newlyParsedParentQueue.showReservationsAsQueues;
+ }
+
+ synchronized void addChildQueue(CSQueue newQueue)
+ throws SchedulerDynamicEditException {
+ if (newQueue.getCapacity() > 0) {
+ throw new SchedulerDynamicEditException("Queue " + newQueue
+ + " being added has non zero capacity.");
+ }
+ boolean added = this.childQueues.add(newQueue);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("updateChildQueues (action: add queue): " + added + " "
+ + getChildQueuesToPrint());
+ }
+ }
+
+ synchronized void removeChildQueue(CSQueue remQueue)
+ throws SchedulerDynamicEditException {
+ if (remQueue.getCapacity() > 0) {
+ throw new SchedulerDynamicEditException("Queue " + remQueue
+ + " being removed has non zero capacity.");
+ }
+ Iterator<CSQueue> qiter = childQueues.iterator();
+ while (qiter.hasNext()) {
+ CSQueue cs = qiter.next();
+ if (cs.equals(remQueue)) {
+ qiter.remove();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Removed child queue: {}", cs.getQueueName());
+ }
+ }
+ }
+ }
+
+ protected synchronized float sumOfChildCapacities() {
+ float ret = 0;
+ for (CSQueue l : childQueues) {
+ ret += l.getCapacity();
+ }
+ return ret;
+ }
+
+ private void updateQuotas(int userLimit, float userLimitFactor,
+ int maxAppsForReservation, int maxAppsPerUserForReservation) {
+ this.userLimit = userLimit;
+ this.userLimitFactor = userLimitFactor;
+ this.maxAppsForReservation = maxAppsForReservation;
+ this.maxAppsPerUserForReservation = maxAppsPerUserForReservation;
+ }
+
+ /**
+ * Number of maximum applications for each of the reservations in this Plan.
+ *
+ * @return maxAppsForreservation
+ */
+ public int getMaxApplicationsForReservations() {
+ return maxAppsForReservation;
+ }
+
+ /**
+ * Number of maximum applications per user for each of the reservations in
+ * this Plan.
+ *
+ * @return maxAppsPerUserForreservation
+ */
+ public int getMaxApplicationsPerUserForReservation() {
+ return maxAppsPerUserForReservation;
+ }
+
+ /**
+ * User limit value for each of the reservations in this Plan.
+ *
+ * @return userLimit
+ */
+ public int getUserLimitForReservation() {
+ return userLimit;
+ }
+
+ /**
+ * User limit factor value for each of the reservations in this Plan.
+ *
+ * @return userLimitFactor
+ */
+ public float getUserLimitFactor() {
+ return userLimitFactor;
+ }
+
+ /**
+ * Determine whether to hide/show the ReservationQueues
+ */
+ public boolean showReservationsAsQueues() {
+ return showReservationsAsQueues;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ReservationQueue.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/capacity/ReservationQueue.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/ReservationQueue.java
new file mode 100644
index 0000000..48733fc
--- /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/capacity/ReservationQueue.java
@@ -0,0 +1,98 @@
+package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
+
+import java.io.IOException;
+
+import org.apache.hadoop.yarn.api.records.Resource;
+import
org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSystem;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerDynamicEditException;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.QueueEntitlement;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This represents a dynamic {@link LeafQueue} managed by the
+ * {@link ReservationSystem}
+ *
+ */
+public class ReservationQueue extends LeafQueue {
+
+ private static final Logger LOG = LoggerFactory
+ .getLogger(ReservationQueue.class);
+
+ private PlanQueue parent;
+
+ private int maxSystemApps;
+
+ public ReservationQueue(CapacitySchedulerContext cs, String queueName,
+ PlanQueue parent) {
+ super(cs, queueName, parent, null);
+ maxSystemApps = cs.getConfiguration().getMaximumSystemApplications();
+ // the following parameters are common to all reservation in the plan
+ updateQuotas(parent.getUserLimitForReservation(),
+ parent.getUserLimitFactor(),
+ parent.getMaxApplicationsForReservations(),
+ parent.getMaxApplicationsPerUserForReservation());
+ this.parent = parent;
+ }
+
+ @Override
+ public synchronized void reinitialize(CSQueue newlyParsedQueue,
+ Resource clusterResource) throws IOException {
+ // Sanity check
+ if (!(newlyParsedQueue instanceof ReservationQueue)
+ || !newlyParsedQueue.getQueuePath().equals(getQueuePath())) {
+ throw new IOException("Trying to reinitialize " + getQueuePath()
+ + " from " + newlyParsedQueue.getQueuePath());
+ }
+ CSQueueUtils.updateQueueStatistics(
+ parent.schedulerContext.getResourceCalculator(), newlyParsedQueue,
+ parent, parent.schedulerContext.getClusterResource(),
+ parent.schedulerContext.getMinimumResourceCapability());
+ super.reinitialize(newlyParsedQueue, clusterResource);
+ updateQuotas(parent.getUserLimitForReservation(),
+ parent.getUserLimitFactor(),
+ parent.getMaxApplicationsForReservations(),
+ parent.getMaxApplicationsPerUserForReservation());
+ }
+
+ /**
+ * This methods to change capacity for a queue and adjusts its
+ * absoluteCapacity
+ *
+ * @param entitlement the new entitlement for the queue (capacity,
+ * maxCapacity, etc..)
+ * @throws SchedulerDynamicEditException
+ */
+ public synchronized void setEntitlement(QueueEntitlement entitlement)
+ throws SchedulerDynamicEditException {
+ float capacity = entitlement.getCapacity();
+ if (capacity < 0 || capacity > 1.0f) {
+ throw new SchedulerDynamicEditException(
+ "Capacity demand is not in the [0,1] range: " + capacity);
+ }
+ setCapacity(capacity);
+ setAbsoluteCapacity(getParent().getAbsoluteCapacity() * getCapacity());
+ setMaxApplications((int) (maxSystemApps * getAbsoluteCapacity()));
+ // note: we currently set maxCapacity to capacity
+ // this might be revised later
+ setMaxCapacity(entitlement.getMaxCapacity());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("successfully changed to " + capacity + " for queue "
+ + this.getQueueName());
+ }
+ }
+
+ private void updateQuotas(int userLimit, float userLimitFactor,
+ int maxAppsForReservation, int maxAppsPerUserForReservation) {
+ setUserLimit(userLimit);
+ setUserLimitFactor(userLimitFactor);
+ setMaxApplications(maxAppsForReservation);
+ maxApplicationsPerUser = maxAppsPerUserForReservation;
+ }
+
+ // used by the super constructor, we initialize to zero
+ protected float getCapacityFromConf() {
+ return 0f;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/QueueEntitlement.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/common/QueueEntitlement.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/QueueEntitlement.java
new file mode 100644
index 0000000..a348e13
--- /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/common/QueueEntitlement.java
@@ -0,0 +1,28 @@
+package org.apache.hadoop.yarn.server.resourcemanager.scheduler.common;
+
+public class QueueEntitlement {
+
+ private float capacity;
+ private float maxCapacity;
+
+ public QueueEntitlement(float capacity, float maxCapacity){
+ this.setCapacity(capacity);
+ this.maxCapacity = maxCapacity;
+ }
+
+ public float getMaxCapacity() {
+ return maxCapacity;
+ }
+
+ public void setMaxCapacity(float maxCapacity) {
+ this.maxCapacity = maxCapacity;
+ }
+
+ public float getCapacity() {
+ return capacity;
+ }
+
+ public void setCapacity(float capacity) {
+ this.capacity = capacity;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/CapacitySchedulerQueueInfo.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/webapp/dao/CapacitySchedulerQueueInfo.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/CapacitySchedulerQueueInfo.java
index ac16ce0..0a8faad 100644
---
a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/CapacitySchedulerQueueInfo.java
+++
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/CapacitySchedulerQueueInfo.java
@@ -25,6 +25,7 @@ import javax.xml.bind.annotation.XmlTransient;
import org.apache.hadoop.yarn.api.records.QueueState;
import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueue;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.PlanQueue;
@XmlRootElement
@XmlAccessorType(XmlAccessType.FIELD)
@@ -48,6 +49,7 @@ public class CapacitySchedulerQueueInfo {
protected QueueState state;
protected CapacitySchedulerQueueInfoList queues;
protected ResourceInfo resourcesUsed;
+ private boolean hideReservationQueues = true;
CapacitySchedulerQueueInfo() {
};
@@ -69,6 +71,10 @@ public class CapacitySchedulerQueueInfo {
queueName = q.getQueueName();
state = q.getState();
resourcesUsed = new ResourceInfo(q.getUsedResources());
+ if(q instanceof PlanQueue &&
+ ((PlanQueue)q).showReservationsAsQueues()) {
+ hideReservationQueues = false;
+ }
}
public float getCapacity() {
@@ -112,6 +118,9 @@ public class CapacitySchedulerQueueInfo {
}
public CapacitySchedulerQueueInfoList getQueues() {
+ if(hideReservationQueues) {
+ return new CapacitySchedulerQueueInfoList();
+ }
return this.queues;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java
----------------------------------------------------------------------
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/TestCapacityScheduler.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/TestCapacityScheduler.java
index e5f4df2..f7c098c 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/TestCapacityScheduler.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/TestCapacityScheduler.java
@@ -410,7 +410,7 @@ public class TestCapacityScheduler {
cs.stop();
}
- private void checkQueueCapacities(CapacityScheduler cs,
+ void checkQueueCapacities(CapacityScheduler cs,
float capacityA, float capacityB) {
CSQueue rootQueue = cs.getRootQueue();
CSQueue queueA = findQueue(rootQueue, A);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacitySchedulerDynamicBehavior.java
----------------------------------------------------------------------
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/TestCapacitySchedulerDynamicBehavior.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/TestCapacitySchedulerDynamicBehavior.java
new file mode 100644
index 0000000..aecbfa8
--- /dev/null
+++
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacitySchedulerDynamicBehavior.java
@@ -0,0 +1,282 @@
+/**
+ * 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.capacity;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerDynamicEditException;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.QueueEntitlement;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestCapacitySchedulerDynamicBehavior {
+ private static final Log LOG = LogFactory
+ .getLog(TestCapacitySchedulerDynamicBehavior.class);
+ private static final String A = CapacitySchedulerConfiguration.ROOT + ".a";
+ private static final String B = CapacitySchedulerConfiguration.ROOT + ".b";
+ private static final String B1 = B + ".b1";
+ private static final String B2 = B + ".b2";
+ private static final String B3 = B + ".b3";
+ private static float A_CAPACITY = 10.5f;
+ private static float B_CAPACITY = 89.5f;
+ private static float A1_CAPACITY = 30;
+ private static float A2_CAPACITY = 70;
+ private static float B1_CAPACITY = 79.2f;
+ private static float B2_CAPACITY = 0.8f;
+ private static float B3_CAPACITY = 20;
+
+ private final TestCapacityScheduler tcs = new TestCapacityScheduler();
+
+ private int GB = 1024;
+
+ private MockRM rm;
+
+ @Before
+ public void setUp() {
+ CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();
+ setupPlanQueueConfiguration(conf);
+ conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
+ ResourceScheduler.class);
+ conf.setBoolean(YarnConfiguration.RM_RESERVATIONS_ENABLE, false);
+ rm = new MockRM(conf);
+ rm.start();
+ }
+
+ @Test
+ public void testRefreshQueuesWithReservations() throws Exception {
+ CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
+
+ // Test add one reservation dynamically and manually modify capacity
+ ReservationQueue a1 =
+ new ReservationQueue(cs, "a1", (PlanQueue) cs.getQueue("a"));
+ cs.addQueue(a1);
+ a1.setEntitlement(new QueueEntitlement(A1_CAPACITY / 100, 1f));
+
+ // Test add another reservation queue and use setEntitlement to modify
+ // capacity
+ ReservationQueue a2 =
+ new ReservationQueue(cs, "a2", (PlanQueue) cs.getQueue("a"));
+ cs.addQueue(a2);
+ cs.setEntitlement("a2", new QueueEntitlement(A2_CAPACITY / 100, 1.0f));
+
+ // Verify all allocations match
+ tcs.checkQueueCapacities(cs, A_CAPACITY, B_CAPACITY);
+
+ // Reinitialize and verify all dynamic queued survived
+ CapacitySchedulerConfiguration conf = cs.getConfiguration();
+ conf.setCapacity(A, 80f);
+ conf.setCapacity(B, 20f);
+ cs.reinitialize(conf, rm.getRMContext());
+
+ tcs.checkQueueCapacities(cs, 80f, 20f);
+ }
+
+ @Test
+ public void testAddQueueFailCases() throws Exception {
+ CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
+
+ try {
+ // Test invalid addition (adding non-zero size queue)
+ ReservationQueue a1 =
+ new ReservationQueue(cs, "a1", (PlanQueue) cs.getQueue("a"));
+ a1.setEntitlement(new QueueEntitlement(A1_CAPACITY / 100, 1f));
+ cs.addQueue(a1);
+ fail();
+ } catch (Exception e) {
+ // expected
+ }
+
+ // Test add one reservation dynamically and manually modify capacity
+ ReservationQueue a1 =
+ new ReservationQueue(cs, "a1", (PlanQueue) cs.getQueue("a"));
+ cs.addQueue(a1);
+ a1.setEntitlement(new QueueEntitlement(A1_CAPACITY / 100, 1f));
+
+ // Test add another reservation queue and use setEntitlement to modify
+ // capacity
+ ReservationQueue a2 =
+ new ReservationQueue(cs, "a2", (PlanQueue) cs.getQueue("a"));
+
+ cs.addQueue(a2);
+
+ try {
+ // Test invalid entitlement (sum of queues exceed 100%)
+ cs.setEntitlement("a2", new QueueEntitlement(A2_CAPACITY / 100 + 0.1f,
+ 1.0f));
+ fail();
+ } catch (Exception e) {
+ // expected
+ }
+
+ cs.setEntitlement("a2", new QueueEntitlement(A2_CAPACITY / 100, 1.0f));
+
+ // Verify all allocations match
+ tcs.checkQueueCapacities(cs, A_CAPACITY, B_CAPACITY);
+
+ cs.stop();
+ }
+
+ @Test
+ public void testRemoveQueue() throws Exception {
+ CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
+
+ // Test add one reservation dynamically and manually modify capacity
+ ReservationQueue a1 =
+ new ReservationQueue(cs, "a1", (PlanQueue) cs.getQueue("a"));
+ cs.addQueue(a1);
+ a1.setEntitlement(new QueueEntitlement(A1_CAPACITY / 100, 1f));
+
+ // submit an app
+ RMApp app = rm.submitApp(GB, "test-move-1", "user_0", null, "a1");
+ // check preconditions
+ List<ApplicationAttemptId> appsInA1 = cs.getAppsInQueue("a1");
+ assertEquals(1, appsInA1.size());
+ try {
+ cs.removeQueue("a1");
+ fail();
+ } catch (SchedulerDynamicEditException s) {
+ // expected a1 contains applications
+ }
+ // clear queue by killling all apps
+ cs.killAllAppsInQueue("a1");
+ // wait for events of move to propagate
+ rm.waitForState(app.getApplicationId(), RMAppState.KILLED);
+
+ try {
+ cs.removeQueue("a1");
+ fail();
+ } catch (SchedulerDynamicEditException s) {
+ // expected a1 is not zero capacity
+ }
+ // set capacity to zero
+ cs.setEntitlement("a1", new QueueEntitlement(0f, 0f));
+ cs.removeQueue("a1");
+
+ assertTrue(cs.getQueue("a1") == null);
+
+ rm.stop();
+ }
+
+ @Test
+ public void testMoveAppToPlanQueue() throws Exception {
+ CapacityScheduler scheduler = (CapacityScheduler)
rm.getResourceScheduler();
+
+ // submit an app
+ RMApp app = rm.submitApp(GB, "test-move-1", "user_0", null, "b1");
+ ApplicationAttemptId appAttemptId =
+ rm.getApplicationReport(app.getApplicationId())
+ .getCurrentApplicationAttemptId();
+
+ // check preconditions
+ List<ApplicationAttemptId> appsInB1 = scheduler.getAppsInQueue("b1");
+ assertEquals(1, appsInB1.size());
+
+ List<ApplicationAttemptId> appsInB = scheduler.getAppsInQueue("b");
+ assertEquals(1, appsInB.size());
+ assertTrue(appsInB.contains(appAttemptId));
+
+ List<ApplicationAttemptId> appsInA = scheduler.getAppsInQueue("a");
+ assertTrue(appsInA.isEmpty());
+
+ String queue =
+ scheduler.getApplicationAttempt(appsInB1.get(0)).getQueue()
+ .getQueueName();
+ Assert.assertTrue(queue.equals("b1"));
+
+ List<ApplicationAttemptId> appsInRoot = scheduler.getAppsInQueue("root");
+ assertTrue(appsInRoot.contains(appAttemptId));
+ assertEquals(1, appsInRoot.size());
+
+ // create the default reservation queue
+ String defQName = "a" + PlanQueue.DEFAULT_QUEUE_SUFFIX;
+ ReservationQueue defQ =
+ new ReservationQueue(scheduler, defQName,
+ (PlanQueue) scheduler.getQueue("a"));
+ scheduler.addQueue(defQ);
+ defQ.setEntitlement(new QueueEntitlement(1f, 1f));
+
+ List<ApplicationAttemptId> appsInDefQ = scheduler.getAppsInQueue(defQName);
+ assertTrue(appsInDefQ.isEmpty());
+
+ // now move the app to plan queue
+ scheduler.moveApplication(app.getApplicationId(), "a");
+
+ // check postconditions
+ appsInDefQ = scheduler.getAppsInQueue(defQName);
+ assertEquals(1, appsInDefQ.size());
+ queue =
+ scheduler.getApplicationAttempt(appsInDefQ.get(0)).getQueue()
+ .getQueueName();
+ Assert.assertTrue(queue.equals(defQName));
+
+ appsInA = scheduler.getAppsInQueue("a");
+ assertTrue(appsInA.contains(appAttemptId));
+ assertEquals(1, appsInA.size());
+
+ appsInRoot = scheduler.getAppsInQueue("root");
+ assertTrue(appsInRoot.contains(appAttemptId));
+ assertEquals(1, appsInRoot.size());
+
+ appsInB1 = scheduler.getAppsInQueue("b1");
+ assertTrue(appsInB1.isEmpty());
+
+ appsInB = scheduler.getAppsInQueue("b");
+ assertTrue(appsInB.isEmpty());
+
+ rm.stop();
+ }
+
+ private void setupPlanQueueConfiguration(CapacitySchedulerConfiguration
conf) {
+
+ conf.setQueues(CapacitySchedulerConfiguration.ROOT,
+ new String[] { "a", "b" });
+
+ conf.setCapacity(A, A_CAPACITY);
+ conf.setCapacity(B, B_CAPACITY);
+
+ // Define 2nd-level queues
+ conf.setQueues(B, new String[] { "b1", "b2", "b3" });
+ conf.setCapacity(B1, B1_CAPACITY);
+ conf.setUserLimitFactor(B1, 100.0f);
+ conf.setCapacity(B2, B2_CAPACITY);
+ conf.setUserLimitFactor(B2, 100.0f);
+ conf.setCapacity(B3, B3_CAPACITY);
+ conf.setUserLimitFactor(B3, 100.0f);
+
+ conf.setReservableQueue(A, true);
+ conf.setReservationWindow(A, 86400 * 1000);
+ conf.setAverageCapacity(A, 1.0f);
+
+ LOG.info("Setup a as a plan queue");
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/eb3e40b8/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestReservationQueue.java
----------------------------------------------------------------------
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/TestReservationQueue.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/TestReservationQueue.java
new file mode 100644
index 0000000..c53b7a9
--- /dev/null
+++
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestReservationQueue.java
@@ -0,0 +1,103 @@
+/**
+ * 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.capacity;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerDynamicEditException;
+import
org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.QueueEntitlement;
+import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
+import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
+import org.apache.hadoop.yarn.util.resource.Resources;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestReservationQueue {
+
+ CapacitySchedulerConfiguration csConf;
+ CapacitySchedulerContext csContext;
+ final static int GB = 1024;
+ private final ResourceCalculator resourceCalculator =
+ new DefaultResourceCalculator();
+ ReservationQueue reservationQueue;
+
+ @Before
+ public void setup() {
+
+ // setup a context / conf
+ csConf = new CapacitySchedulerConfiguration();
+ YarnConfiguration conf = new YarnConfiguration();
+ csContext = mock(CapacitySchedulerContext.class);
+ when(csContext.getConfiguration()).thenReturn(csConf);
+ when(csContext.getConf()).thenReturn(conf);
+ when(csContext.getMinimumResourceCapability()).thenReturn(
+ Resources.createResource(GB, 1));
+ when(csContext.getMaximumResourceCapability()).thenReturn(
+ Resources.createResource(16 * GB, 32));
+ when(csContext.getClusterResource()).thenReturn(
+ Resources.createResource(100 * 16 * GB, 100 * 32));
+ when(csContext.getResourceCalculator()).thenReturn(resourceCalculator);
+
+ // create a queue
+ PlanQueue pq = new PlanQueue(csContext, "root", null, null);
+ reservationQueue = new ReservationQueue(csContext, "a", pq);
+
+ }
+
+ @Test
+ public void testAddSubtractCapacity() throws Exception {
+
+ // verify that setting, adding, subtracting capacity works
+ reservationQueue.setCapacity(1.0F);
+ assertTrue(" actual capacity: " + reservationQueue.getCapacity(),
+ reservationQueue.getCapacity() - 1 < CSQueueUtils.EPSILON);
+ reservationQueue.setEntitlement(new QueueEntitlement(0.9f, 1f));
+ assertTrue(" actual capacity: " + reservationQueue.getCapacity(),
+ reservationQueue.getCapacity() - 0.9 < CSQueueUtils.EPSILON);
+ reservationQueue.setEntitlement(new QueueEntitlement(1f, 1f));
+ assertTrue(" actual capacity: " + reservationQueue.getCapacity(),
+ reservationQueue.getCapacity() - 1 < CSQueueUtils.EPSILON);
+ reservationQueue.setEntitlement(new QueueEntitlement(0f, 1f));
+ assertTrue(" actual capacity: " + reservationQueue.getCapacity(),
+ reservationQueue.getCapacity() < CSQueueUtils.EPSILON);
+
+ try {
+ reservationQueue.setEntitlement(new QueueEntitlement(1.1f, 1f));
+ fail();
+ } catch (SchedulerDynamicEditException iae) {
+ // expected
+ assertTrue(" actual capacity: " + reservationQueue.getCapacity(),
+ reservationQueue.getCapacity() - 1 < CSQueueUtils.EPSILON);
+ }
+
+ try {
+ reservationQueue.setEntitlement(new QueueEntitlement(-0.1f, 1f));
+ fail();
+ } catch (SchedulerDynamicEditException iae) {
+ // expected
+ assertTrue(" actual capacity: " + reservationQueue.getCapacity(),
+ reservationQueue.getCapacity() - 1 < CSQueueUtils.EPSILON);
+ }
+
+ }
+}
