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 ha...@apache.org on 2018/04/10 17:33:28 UTC
[31/50] [abbrv] hadoop git commit: YARN-7574. Add support for Node
Labels on Auto Created Leaf Queue Template. Contributed by Suma Shivaprasad.
YARN-7574. Add support for Node Labels on Auto Created Leaf Queue Template. Contributed by Suma Shivaprasad.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/821b0de4
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/821b0de4
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/821b0de4
Branch: refs/heads/YARN-1011
Commit: 821b0de4c59156d4a65112de03ba3e7e1c88e309
Parents: 5700556
Author: Sunil G <su...@apache.org>
Authored: Mon Apr 9 21:17:22 2018 +0530
Committer: Sunil G <su...@apache.org>
Committed: Mon Apr 9 21:17:22 2018 +0530
----------------------------------------------------------------------
.../server/resourcemanager/RMServerUtils.java | 5 +-
.../rmapp/attempt/RMAppAttemptImpl.java | 47 ++
.../resourcemanager/scheduler/Allocation.java | 12 +
.../scheduler/SchedulerUtils.java | 33 +-
.../capacity/AutoCreatedLeafQueue.java | 3 +-
.../AutoCreatedQueueManagementPolicy.java | 12 +-
.../scheduler/capacity/CapacityScheduler.java | 2 +
.../CapacitySchedulerConfiguration.java | 28 +
.../scheduler/capacity/LeafQueue.java | 11 +
.../scheduler/capacity/ManagedParentQueue.java | 5 +-
.../GuaranteedOrZeroCapacityOverTimePolicy.java | 573 +++++++++++--------
.../placement/PendingAskUpdateResult.java | 8 +
.../yarn/server/resourcemanager/MockNM.java | 15 +
.../server/resourcemanager/TestAppManager.java | 20 +-
...stCapacitySchedulerAutoCreatedQueueBase.java | 241 +++++---
.../TestCapacitySchedulerAutoQueueCreation.java | 233 +++++---
.../TestQueueManagementDynamicEditPolicy.java | 30 +-
17 files changed, 834 insertions(+), 444 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMServerUtils.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/RMServerUtils.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMServerUtils.java
index 33451295..ab6bbcf 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMServerUtils.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMServerUtils.java
@@ -236,13 +236,14 @@ public class RMServerUtils {
*/
public static void normalizeAndValidateRequests(List<ResourceRequest> ask,
Resource maximumResource, String queueName, YarnScheduler scheduler,
- RMContext rmContext)
- throws InvalidResourceRequestException {
+ RMContext rmContext) throws InvalidResourceRequestException {
// Get queue from scheduler
QueueInfo queueInfo = null;
try {
queueInfo = scheduler.getQueueInfo(queueName, false, false);
} catch (IOException e) {
+ //Queue may not exist since it could be auto-created in case of
+ // dynamic queues
}
for (ResourceRequest resReq : ask) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.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/rmapp/attempt/RMAppAttemptImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
index c23b135..1b1e2c4 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
@@ -57,6 +57,7 @@ import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.QueueInfo;
import org.apache.hadoop.yarn.api.records.ResourceBlacklistRequest;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
import org.apache.hadoop.yarn.api.records.YarnApplicationAttemptState;
@@ -75,6 +76,9 @@ import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.AMLauncherEvent;
import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.AMLauncherEventType;
import org.apache.hadoop.yarn.server.resourcemanager.blacklist.BlacklistManager;
import org.apache.hadoop.yarn.server.resourcemanager.blacklist.DisabledBlacklistManager;
+
+import org.apache.hadoop.yarn.server.resourcemanager.nodelabels
+ .RMNodeLabelsManager;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.Recoverable;
@@ -1109,6 +1113,49 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
amBlacklist.getBlacklistAdditions() + ") and removals(" +
amBlacklist.getBlacklistRemovals() + ")");
}
+
+ QueueInfo queueInfo = null;
+ for (ResourceRequest amReq : appAttempt.amReqs) {
+ if (amReq.getNodeLabelExpression() == null && ResourceRequest.ANY
+ .equals(amReq.getResourceName())) {
+ String queue = appAttempt.rmApp.getQueue();
+
+ //Load queue only once since queue will be same across attempts
+ if (queueInfo == null) {
+ try {
+ queueInfo = appAttempt.scheduler.getQueueInfo(queue, false,
+ false);
+ } catch (IOException e) {
+ LOG.error("Could not find queue for application : ", e);
+ // Set application status to REJECTED since we cant find the
+ // queue
+ appAttempt.rmContext.getDispatcher().getEventHandler().handle(
+ new RMAppAttemptEvent(appAttempt.getAppAttemptId(),
+ RMAppAttemptEventType.FAIL,
+ "Could not find queue for application : " +
+ appAttempt.rmApp.getQueue()));
+ appAttempt.rmContext.getDispatcher().getEventHandler().handle(
+ new RMAppEvent(appAttempt.rmApp.getApplicationId(), RMAppEventType
+ .APP_REJECTED,
+ "Could not find queue for application : " +
+ appAttempt.rmApp.getQueue()));
+ return RMAppAttemptState.FAILED;
+ }
+ }
+
+ String labelExp = RMNodeLabelsManager.NO_LABEL;
+ if (queueInfo != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Setting default node label expression : " + queueInfo
+ .getDefaultNodeLabelExpression());
+ }
+ labelExp = queueInfo.getDefaultNodeLabelExpression();
+ }
+
+ amReq.setNodeLabelExpression(labelExp);
+ }
+ }
+
// AM resource has been checked when submission
Allocation amContainerAllocation =
appAttempt.scheduler.allocate(
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Allocation.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/Allocation.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Allocation.java
index 768afde..9573ac8 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Allocation.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Allocation.java
@@ -132,4 +132,16 @@ public class Allocation {
public void setResourceLimit(Resource resource) {
this.resourceLimit = resource;
}
+
+ @Override
+ public String toString() {
+ return "Allocation{" + "containers=" + containers + ", strictContainers="
+ + strictContainers + ", fungibleContainers=" + fungibleContainers
+ + ", fungibleResources=" + fungibleResources + ", nmTokens=" + nmTokens
+ + ", increasedContainers=" + increasedContainers
+ + ", decreasedContainers=" + decreasedContainers
+ + ", promotedContainers=" + promotedContainers + ", demotedContainers="
+ + demotedContainers + ", previousAttemptContainers="
+ + previousAttemptContainers + ", resourceLimit=" + resourceLimit + '}';
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.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/SchedulerUtils.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java
index 27563d6..c0d7d86 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java
@@ -186,19 +186,33 @@ public class SchedulerUtils {
ResourceRequest resReq, QueueInfo queueInfo) {
String labelExp = resReq.getNodeLabelExpression();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Requested Node Label Expression : " + labelExp);
+ LOG.debug("Queue Info : " + queueInfo);
+ }
// if queue has default label expression, and RR doesn't have, use the
// default label expression of queue
if (labelExp == null && queueInfo != null && ResourceRequest.ANY
.equals(resReq.getResourceName())) {
+ if ( LOG.isDebugEnabled()) {
+ LOG.debug("Setting default node label expression : " + queueInfo
+ .getDefaultNodeLabelExpression());
+ }
labelExp = queueInfo.getDefaultNodeLabelExpression();
}
- // If labelExp still equals to null, set it to be NO_LABEL
- if (labelExp == null) {
+ // If labelExp still equals to null, it could either be a dynamic queue
+ // or the label is not configured
+ // set it to be NO_LABEL in case of a pre-configured queue. Dynamic
+ // queues are handled in RMAppAttemptImp.ScheduledTransition
+ if (labelExp == null && queueInfo != null) {
labelExp = RMNodeLabelsManager.NO_LABEL;
}
- resReq.setNodeLabelExpression(labelExp);
+
+ if ( labelExp != null) {
+ resReq.setNodeLabelExpression(labelExp);
+ }
}
public static void normalizeAndValidateRequest(ResourceRequest resReq,
@@ -209,6 +223,7 @@ public class SchedulerUtils {
isRecovery, rmContext, null);
}
+
public static void normalizeAndValidateRequest(ResourceRequest resReq,
Resource maximumResource, String queueName, YarnScheduler scheduler,
boolean isRecovery, RMContext rmContext, QueueInfo queueInfo)
@@ -233,11 +248,12 @@ public class SchedulerUtils {
try {
queueInfo = scheduler.getQueueInfo(queueName, false, false);
} catch (IOException e) {
- // it is possible queue cannot get when queue mapping is set, just ignore
- // the queueInfo here, and move forward
+ //Queue may not exist since it could be auto-created in case of
+ // dynamic queues
}
}
SchedulerUtils.normalizeNodeLabelExpressionInRequest(resReq, queueInfo);
+
if (!isRecovery) {
validateResourceRequest(resReq, maximumResource, queueInfo, rmContext);
}
@@ -245,8 +261,7 @@ public class SchedulerUtils {
public static void normalizeAndvalidateRequest(ResourceRequest resReq,
Resource maximumResource, String queueName, YarnScheduler scheduler,
- RMContext rmContext)
- throws InvalidResourceRequestException {
+ RMContext rmContext) throws InvalidResourceRequestException {
normalizeAndvalidateRequest(resReq, maximumResource, queueName, scheduler,
rmContext, null);
}
@@ -296,7 +311,7 @@ public class SchedulerUtils {
+ "resource request has resource name = "
+ resReq.getResourceName());
}
-
+
// we don't allow specify label expression with more than one node labels now
if (labelExp != null && labelExp.contains("&&")) {
throw new InvalidLabelResourceRequestException(
@@ -305,7 +320,7 @@ public class SchedulerUtils {
+ "in a node label expression, node label expression = "
+ labelExp);
}
-
+
if (labelExp != null && !labelExp.trim().isEmpty() && queueInfo != null) {
if (!checkQueueLabelExpression(queueInfo.getAccessibleNodeLabels(),
labelExp, rmContext)) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/AutoCreatedLeafQueue.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/AutoCreatedLeafQueue.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/AutoCreatedLeafQueue.java
index 8b67087..e12b55e 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/AutoCreatedLeafQueue.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/AutoCreatedLeafQueue.java
@@ -148,11 +148,10 @@ public class AutoCreatedLeafQueue extends AbstractAutoCreatedLeafQueue {
try {
for( String nodeLabel : parent.getQueueCapacities().getExistingNodeLabels
()) {
- //TODO - update to use getMaximumCapacity(nodeLabel) in YARN-7574
setEntitlement(nodeLabel, new QueueEntitlement(0.0f,
parent.getLeafQueueTemplate()
.getQueueCapacities()
- .getMaximumCapacity()));
+ .getMaximumCapacity(nodeLabel)));
}
} catch (SchedulerDynamicEditException e) {
throw new IOException(e);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/AutoCreatedQueueManagementPolicy.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/AutoCreatedQueueManagementPolicy.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/AutoCreatedQueueManagementPolicy.java
index f7a4bbd..388e9d6 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/AutoCreatedQueueManagementPolicy.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/AutoCreatedQueueManagementPolicy.java
@@ -18,6 +18,9 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerDynamicEditException;
+
+
+import java.io.IOException;
import java.util.List;
public interface AutoCreatedQueueManagementPolicy {
@@ -26,14 +29,15 @@ public interface AutoCreatedQueueManagementPolicy {
* Initialize policy
* @param schedulerContext Capacity Scheduler context
*/
- void init(CapacitySchedulerContext schedulerContext, ParentQueue parentQueue);
+ void init(CapacitySchedulerContext schedulerContext, ParentQueue
+ parentQueue) throws IOException;
/**
* Reinitialize policy state ( if required )
* @param schedulerContext Capacity Scheduler context
*/
void reinitialize(CapacitySchedulerContext schedulerContext,
- ParentQueue parentQueue);
+ ParentQueue parentQueue) throws IOException;
/**
* Get initial template for the specified leaf queue
@@ -48,6 +52,10 @@ public interface AutoCreatedQueueManagementPolicy {
/**
* Compute/Adjust child queue capacities
* for auto created leaf queues
+ * This computes queue entitlements but does not update LeafQueueState or
+ * queue capacities. Scheduler calls commitQueueManagemetChanges after
+ * validation after applying queue changes and commits to LeafQueueState
+ * are done in commitQueueManagementChanges.
*
* @return returns a list of suggested QueueEntitlementChange(s) which may
* or may not be be enforced by the scheduler
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/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 e59bdde..776e512 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
@@ -1204,6 +1204,8 @@ public class CapacityScheduler extends
updateDemandForQueue.getOrderingPolicy().demandUpdated(application);
}
+ LOG.info("Allocation for application " + applicationAttemptId + " : " +
+ allocation + " with cluster resource : " + getClusterResource());
return allocation;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java
index 1870aef..c41bd96 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java
@@ -1878,6 +1878,15 @@ public class CapacitySchedulerConfiguration extends ReservationSchedulerConfigur
setCapacity(leafQueueConfPrefix, val);
}
+ @VisibleForTesting
+ @Private
+ public void setAutoCreatedLeafQueueTemplateCapacityByLabel(String queuePath,
+ String label, float val) {
+ String leafQueueConfPrefix = getAutoCreatedQueueTemplateConfPrefix(
+ queuePath);
+ setCapacityByLabel(leafQueueConfPrefix, label, val);
+ }
+
@Private
@VisibleForTesting
public void setAutoCreatedLeafQueueConfigMaxCapacity(String queuePath,
@@ -1887,6 +1896,15 @@ public class CapacitySchedulerConfiguration extends ReservationSchedulerConfigur
setMaximumCapacity(leafQueueConfPrefix, val);
}
+ @Private
+ @VisibleForTesting
+ public void setAutoCreatedLeafQueueTemplateMaxCapacity(String queuePath,
+ String label, float val) {
+ String leafQueueConfPrefix = getAutoCreatedQueueTemplateConfPrefix(
+ queuePath);
+ setMaximumCapacityByLabel(leafQueueConfPrefix, label, val);
+ }
+
@VisibleForTesting
@Private
public void setAutoCreatedLeafQueueConfigUserLimit(String queuePath,
@@ -1905,6 +1923,16 @@ public class CapacitySchedulerConfiguration extends ReservationSchedulerConfigur
setUserLimitFactor(leafQueueConfPrefix, val);
}
+ @Private
+ @VisibleForTesting
+ public void setAutoCreatedLeafQueueConfigDefaultNodeLabelExpression(String
+ queuePath,
+ String expression) {
+ String leafQueueConfPrefix = getAutoCreatedQueueTemplateConfPrefix(
+ queuePath);
+ setDefaultNodeLabelExpression(leafQueueConfPrefix, expression);
+ }
+
public static String getUnits(String resourceValue) {
String units;
for (int i = 0; i < resourceValue.length(); i++) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/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 8d1428d..1ae8f91 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
@@ -778,6 +778,17 @@ public class LeafQueue extends AbstractCSQueue {
metrics.setAMResouceLimit(nodePartition, amResouceLimit);
queueUsage.setAMLimit(nodePartition, amResouceLimit);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Queue: " + getQueueName() + ", node label : " +
+ nodePartition
+ + ", queue "
+ + "partition "
+ + "resource : " + queuePartitionResource + ','
+ + " queue current limit : " + queueCurrentLimit + ","
+ + " queue partition usable resource : "
+ + queuePartitionUsableResource + ","
+ + " amResourceLimit : " + amResouceLimit);
+ }
return amResouceLimit;
} finally {
writeLock.unlock();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ManagedParentQueue.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/ManagedParentQueue.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/ManagedParentQueue.java
index cbdb21d..2494000 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/ManagedParentQueue.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/ManagedParentQueue.java
@@ -132,7 +132,7 @@ public class ManagedParentQueue extends AbstractManagedParentQueue {
}
}
- private void initializeQueueManagementPolicy() {
+ private void initializeQueueManagementPolicy() throws IOException {
queueManagementPolicy =
csContext.getConfiguration().getAutoCreatedQueueManagementPolicyClass(
getQueuePath());
@@ -140,7 +140,7 @@ public class ManagedParentQueue extends AbstractManagedParentQueue {
queueManagementPolicy.init(csContext, this);
}
- private void reinitializeQueueManagementPolicy() {
+ private void reinitializeQueueManagementPolicy() throws IOException {
AutoCreatedQueueManagementPolicy managementPolicy =
csContext.getConfiguration().getAutoCreatedQueueManagementPolicyClass(
getQueuePath());
@@ -339,6 +339,7 @@ public class ManagedParentQueue extends AbstractManagedParentQueue {
((AutoCreatedLeafQueue) childQueue).validateConfigurations(template);
break;
}
+
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/queuemanagement/GuaranteedOrZeroCapacityOverTimePolicy.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/queuemanagement/GuaranteedOrZeroCapacityOverTimePolicy.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/queuemanagement/GuaranteedOrZeroCapacityOverTimePolicy.java
index aee6405..b2301fd 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/queuemanagement/GuaranteedOrZeroCapacityOverTimePolicy.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/queuemanagement/GuaranteedOrZeroCapacityOverTimePolicy.java
@@ -50,6 +50,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica
import org.apache.hadoop.yarn.util.Clock;
import org.apache.hadoop.yarn.util.MonotonicClock;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
@@ -63,8 +64,6 @@ import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-import static org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager
- .NO_LABEL;
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler
.capacity.CSQueueUtils.EPSILON;
@@ -85,8 +84,6 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
private static final Log LOG = LogFactory.getLog(
GuaranteedOrZeroCapacityOverTimePolicy.class);
- private AutoCreatedLeafQueueConfig ZERO_CAPACITY_ENTITLEMENT;
-
private ReentrantReadWriteLock.WriteLock writeLock;
private ReentrantReadWriteLock.ReadLock readLock;
@@ -97,12 +94,70 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
private QueueCapacities leafQueueTemplateCapacities;
- private Map<String, LeafQueueState> leafQueueStateMap = new HashMap<>();
+ private Set<String> leafQueueTemplateNodeLabels;
+
+ private LeafQueueState leafQueueState = new LeafQueueState();
private Clock clock = new MonotonicClock();
private class LeafQueueState {
+ //map of partition-> queueName->{leaf queue's state}
+ private Map<String, Map<String, LeafQueueStatePerPartition>>
+ leafQueueStateMap = new HashMap<>();
+
+ public boolean containsLeafQueue(String leafQueueName, String partition) {
+ if (leafQueueStateMap.containsKey(partition)) {
+ return leafQueueStateMap.get(partition).containsKey(leafQueueName);
+ }
+ return false;
+ }
+
+ private boolean containsPartition(String partition) {
+ if (leafQueueStateMap.containsKey(partition)) {
+ return true;
+ }
+ return false;
+ }
+
+ private boolean addLeafQueueStateIfNotExists(String leafQueueName,
+ String partition, LeafQueueStatePerPartition leafQueueState) {
+ if (!containsPartition(partition)) {
+ leafQueueStateMap.put(partition, new HashMap<>());
+ }
+ if (!containsLeafQueue(leafQueueName, partition)) {
+ leafQueueStateMap.get(partition).put(leafQueueName, leafQueueState);
+ return true;
+ }
+ return false;
+ }
+
+ public boolean createLeafQueueStateIfNotExists(LeafQueue leafQueue,
+ String partition) {
+ return addLeafQueueStateIfNotExists(leafQueue.getQueueName(), partition,
+ new LeafQueueStatePerPartition());
+ }
+
+ public LeafQueueStatePerPartition getLeafQueueStatePerPartition(
+ String leafQueueName, String partition) {
+ if (leafQueueStateMap.get(partition) != null) {
+ return leafQueueStateMap.get(partition).get(leafQueueName);
+ }
+ return null;
+ }
+
+ public Map<String, Map<String, LeafQueueStatePerPartition>>
+ getLeafQueueStateMap() {
+ return leafQueueStateMap;
+ }
+
+ private void clear() {
+ leafQueueStateMap.clear();
+ }
+ }
+
+ private class LeafQueueStatePerPartition {
+
private AtomicBoolean isActive = new AtomicBoolean(false);
private long mostRecentActivationTime;
@@ -139,41 +194,16 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
}
}
- private boolean containsLeafQueue(String leafQueueName) {
- return leafQueueStateMap.containsKey(leafQueueName);
- }
-
- private boolean addLeafQueueStateIfNotExists(String leafQueueName,
- LeafQueueState leafQueueState) {
- if (!containsLeafQueue(leafQueueName)) {
- leafQueueStateMap.put(leafQueueName, leafQueueState);
- return true;
- }
- return false;
- }
-
- private boolean addLeafQueueStateIfNotExists(LeafQueue leafQueue) {
- return addLeafQueueStateIfNotExists(leafQueue.getQueueName(),
- new LeafQueueState());
- }
-
- private void clearLeafQueueState() {
- leafQueueStateMap.clear();
- }
-
private class ParentQueueState {
private Map<String, Float> totalAbsoluteActivatedChildQueueCapacityByLabel =
new HashMap<String, Float>();
- private float getAbsoluteActivatedChildQueueCapacity() {
- return getAbsoluteActivatedChildQueueCapacity(NO_LABEL);
- }
-
private float getAbsoluteActivatedChildQueueCapacity(String nodeLabel) {
try {
readLock.lock();
- Float totalActivatedCapacity = getByLabel(nodeLabel);
+ Float totalActivatedCapacity = getAbsActivatedChildQueueCapacityByLabel(
+ nodeLabel);
if (totalActivatedCapacity != null) {
return totalActivatedCapacity;
} else{
@@ -188,11 +218,14 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
float childQueueCapacity) {
try {
writeLock.lock();
- Float activatedChildCapacity = getByLabel(nodeLabel);
+ Float activatedChildCapacity = getAbsActivatedChildQueueCapacityByLabel(
+ nodeLabel);
if (activatedChildCapacity != null) {
- setByLabel(nodeLabel, activatedChildCapacity + childQueueCapacity);
+ setAbsActivatedChildQueueCapacityByLabel(nodeLabel,
+ activatedChildCapacity + childQueueCapacity);
} else{
- setByLabel(nodeLabel, childQueueCapacity);
+ setAbsActivatedChildQueueCapacityByLabel(nodeLabel,
+ childQueueCapacity);
}
} finally {
writeLock.unlock();
@@ -203,22 +236,25 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
float childQueueCapacity) {
try {
writeLock.lock();
- Float activatedChildCapacity = getByLabel(nodeLabel);
+ Float activatedChildCapacity = getAbsActivatedChildQueueCapacityByLabel(
+ nodeLabel);
if (activatedChildCapacity != null) {
- setByLabel(nodeLabel, activatedChildCapacity - childQueueCapacity);
+ setAbsActivatedChildQueueCapacityByLabel(nodeLabel,
+ activatedChildCapacity - childQueueCapacity);
} else{
- setByLabel(nodeLabel, childQueueCapacity);
+ setAbsActivatedChildQueueCapacityByLabel(nodeLabel,
+ childQueueCapacity);
}
} finally {
writeLock.unlock();
}
}
- Float getByLabel(String label) {
+ Float getAbsActivatedChildQueueCapacityByLabel(String label) {
return totalAbsoluteActivatedChildQueueCapacityByLabel.get(label);
}
- Float setByLabel(String label, float val) {
+ Float setAbsActivatedChildQueueCapacityByLabel(String label, float val) {
return totalAbsoluteActivatedChildQueueCapacityByLabel.put(label, val);
}
@@ -256,13 +292,12 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
@Override
public void init(final CapacitySchedulerContext schedulerContext,
- final ParentQueue parentQueue) {
+ final ParentQueue parentQueue) throws IOException {
this.scheduler = schedulerContext;
ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
readLock = lock.readLock();
writeLock = lock.writeLock();
-
if (!(parentQueue instanceof ManagedParentQueue)) {
throw new IllegalArgumentException(
"Expected instance of type " + ManagedParentQueue.class);
@@ -278,15 +313,43 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
+ leafQueueTemplate.getQueueCapacities() + "]");
}
- private void initializeLeafQueueTemplate(ManagedParentQueue parentQueue) {
+ private void initializeLeafQueueTemplate(ManagedParentQueue parentQueue)
+ throws IOException {
leafQueueTemplate = parentQueue.getLeafQueueTemplate();
leafQueueTemplateCapacities = leafQueueTemplate.getQueueCapacities();
- ZERO_CAPACITY_ENTITLEMENT = buildTemplate(0.0f,
- leafQueueTemplateCapacities.getMaximumCapacity());
+ Set<String> parentQueueLabels = parentQueue.getNodeLabelsForQueue();
+ for (String nodeLabel : leafQueueTemplateCapacities
+ .getExistingNodeLabels()) {
+
+ if (!parentQueueLabels.contains(nodeLabel)) {
+ LOG.error("Invalid node label " + nodeLabel
+ + " on configured leaf template on parent" + " queue " + parentQueue
+ .getQueueName());
+ throw new IOException("Invalid node label " + nodeLabel
+ + " on configured leaf template on parent" + " queue " + parentQueue
+ .getQueueName());
+ }
+ }
+
+ leafQueueTemplateNodeLabels =
+ leafQueueTemplateCapacities.getExistingNodeLabels();
+
}
+ /**
+ * Compute/Adjust child queue capacities
+ * for auto created leaf queues
+ * This computes queue entitlements but does not update LeafQueueState or
+ * queue capacities. Scheduler calls commitQueueManagemetChanges after
+ * validation after applying queue changes and commits to LeafQueueState
+ * are done in commitQueueManagementChanges.
+ *
+ * @return List of Queue Management change suggestions which could potentially
+ * be committed/rejected by the scheduler due to validation failures
+ * @throws SchedulerDynamicEditException
+ */
@Override
public List<QueueManagementChange> computeQueueManagementChanges()
throws SchedulerDynamicEditException {
@@ -298,70 +361,92 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
try {
readLock.lock();
List<QueueManagementChange> queueManagementChanges = new ArrayList<>();
+ List<FiCaSchedulerApp> pendingApps = getSortedPendingApplications();
+
+ //Map of LeafQueue->QueueCapacities - keep adding the computed
+ // entitlements to this map and finally
+ // build the leaf queue configuration Template for all identified leaf
+ // queues
+ Map<String, QueueCapacities> leafQueueEntitlements = new HashMap<>();
+ for (String nodeLabel : leafQueueTemplateNodeLabels) {
+ // check if any leaf queues need to be deactivated based on pending
+ // applications
+ float parentAbsoluteCapacity =
+ managedParentQueue.getQueueCapacities().getAbsoluteCapacity(
+ nodeLabel);
+ float leafQueueTemplateAbsoluteCapacity =
+ leafQueueTemplateCapacities.getAbsoluteCapacity(nodeLabel);
+ Map<String, QueueCapacities> deactivatedLeafQueues =
+ deactivateLeafQueuesIfInActive(managedParentQueue, nodeLabel,
+ leafQueueEntitlements);
+
+ float deactivatedCapacity = getTotalDeactivatedCapacity(
+ deactivatedLeafQueues, nodeLabel);
+
+ float sumOfChildQueueActivatedCapacity = parentQueueState.
+ getAbsoluteActivatedChildQueueCapacity(nodeLabel);
+
+ //Check if we need to activate anything at all?
+ float availableCapacity =
+ parentAbsoluteCapacity - sumOfChildQueueActivatedCapacity
+ + deactivatedCapacity + EPSILON;
- // check if any leaf queues need to be deactivated based on pending
- // applications and
- float parentAbsoluteCapacity =
- managedParentQueue.getQueueCapacities().getAbsoluteCapacity();
-
- float leafQueueTemplateAbsoluteCapacity =
- leafQueueTemplateCapacities.getAbsoluteCapacity();
- Map<String, QueueCapacities> deactivatedLeafQueues =
- deactivateLeafQueuesIfInActive(managedParentQueue, queueManagementChanges);
-
- float deactivatedCapacity = getTotalDeactivatedCapacity(
- deactivatedLeafQueues);
-
- float sumOfChildQueueActivatedCapacity = parentQueueState.
- getAbsoluteActivatedChildQueueCapacity();
-
- //Check if we need to activate anything at all?
- float availableCapacity = getAvailableCapacity(parentAbsoluteCapacity,
- deactivatedCapacity, sumOfChildQueueActivatedCapacity);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Parent queue : " + managedParentQueue.getQueueName() + " absCapacity = "
- + parentAbsoluteCapacity + ", leafQueueAbsoluteCapacity = "
- + leafQueueTemplateAbsoluteCapacity + ", deactivatedCapacity = "
- + deactivatedCapacity + " , absChildActivatedCapacity = "
- + sumOfChildQueueActivatedCapacity + ", availableCapacity = "
- + availableCapacity);
- }
-
- if (availableCapacity >= leafQueueTemplateAbsoluteCapacity) {
- //sort applications across leaf queues by submit time
- List<FiCaSchedulerApp> pendingApps = getSortedPendingApplications();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Parent queue : " + managedParentQueue.getQueueName()
+ + ", nodeLabel = " + nodeLabel + ", absCapacity = "
+ + parentAbsoluteCapacity + ", leafQueueAbsoluteCapacity = "
+ + leafQueueTemplateAbsoluteCapacity + ", deactivatedCapacity = "
+ + deactivatedCapacity + " , absChildActivatedCapacity = "
+ + sumOfChildQueueActivatedCapacity + ", availableCapacity = "
+ + availableCapacity);
+ }
- if (pendingApps.size() > 0) {
- int maxLeafQueuesTobeActivated = getMaxLeavesToBeActivated(
- availableCapacity, leafQueueTemplateAbsoluteCapacity,
- pendingApps.size());
+ if (availableCapacity >= leafQueueTemplateAbsoluteCapacity) {
+ //sort applications across leaf queues by submit time
+ if (pendingApps.size() > 0) {
+ int maxLeafQueuesTobeActivated = getMaxLeavesToBeActivated(
+ availableCapacity, leafQueueTemplateAbsoluteCapacity,
+ pendingApps.size());
- if (LOG.isDebugEnabled()) {
- LOG.debug("Found " + maxLeafQueuesTobeActivated
- + " leaf queues to be activated with " + pendingApps.size()
- + " apps ");
- }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Found " + maxLeafQueuesTobeActivated + " leaf queues"
+ + " to be activated with " + pendingApps.size() + " apps ");
+ }
- LinkedHashSet<String> leafQueuesToBeActivated = getSortedLeafQueues(
- pendingApps, maxLeafQueuesTobeActivated,
- deactivatedLeafQueues.keySet());
+ LinkedHashSet<String> leafQueuesToBeActivated = getSortedLeafQueues(
+ nodeLabel, pendingApps, maxLeafQueuesTobeActivated,
+ deactivatedLeafQueues.keySet());
- //Compute entitlement changes for the identified leaf queues
- // which is appended to the List of queueManagementChanges
- computeQueueManagementChanges(leafQueuesToBeActivated,
- queueManagementChanges, availableCapacity,
- leafQueueTemplateAbsoluteCapacity);
+ //Compute entitlement changes for the identified leaf queues
+ // which is appended to the List of computedEntitlements
+ updateLeafQueueCapacitiesByLabel(nodeLabel, leafQueuesToBeActivated,
+ leafQueueEntitlements);
- if (LOG.isDebugEnabled()) {
- if (leafQueuesToBeActivated.size() > 0) {
- LOG.debug(
- "Activated leaf queues : [" + leafQueuesToBeActivated + "]");
+ if (LOG.isDebugEnabled()) {
+ if (leafQueuesToBeActivated.size() > 0) {
+ LOG.debug("Activated leaf queues : [" + leafQueuesToBeActivated
+ + "]");
+ }
}
}
}
}
+
+ //Populate new entitlements
+
+ for (final Iterator<Map.Entry<String, QueueCapacities>> iterator =
+ leafQueueEntitlements.entrySet().iterator(); iterator.hasNext(); ) {
+ Map.Entry<String, QueueCapacities> queueCapacities = iterator.next();
+ String leafQueueName = queueCapacities.getKey();
+ AutoCreatedLeafQueue leafQueue =
+ (AutoCreatedLeafQueue) scheduler.getCapacitySchedulerQueueManager()
+ .getQueue(leafQueueName);
+ AutoCreatedLeafQueueConfig newTemplate = buildTemplate(
+ queueCapacities.getValue());
+ queueManagementChanges.add(
+ new QueueManagementChange.UpdateQueue(leafQueue, newTemplate));
+
+ }
return queueManagementChanges;
} finally {
readLock.unlock();
@@ -369,14 +454,14 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
}
private float getTotalDeactivatedCapacity(
- Map<String, QueueCapacities> deactivatedLeafQueues) {
+ Map<String, QueueCapacities> deactivatedLeafQueues, String nodeLabel) {
float deactivatedCapacity = 0;
for (Iterator<Map.Entry<String, QueueCapacities>> iterator =
deactivatedLeafQueues.entrySet().iterator(); iterator.hasNext(); ) {
Map.Entry<String, QueueCapacities> deactivatedQueueCapacity =
iterator.next();
deactivatedCapacity +=
- deactivatedQueueCapacity.getValue().getAbsoluteCapacity();
+ deactivatedQueueCapacity.getValue().getAbsoluteCapacity(nodeLabel);
}
return deactivatedCapacity;
}
@@ -385,20 +470,42 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
void updateLeafQueueState() {
try {
writeLock.lock();
+ Set<String> newPartitions = new HashSet<>();
Set<String> newQueues = new HashSet<>();
+
for (CSQueue newQueue : managedParentQueue.getChildQueues()) {
if (newQueue instanceof LeafQueue) {
- addLeafQueueStateIfNotExists((LeafQueue) newQueue);
+ for (String nodeLabel : leafQueueTemplateNodeLabels) {
+ leafQueueState.createLeafQueueStateIfNotExists((LeafQueue) newQueue,
+ nodeLabel);
+ newPartitions.add(nodeLabel);
+ }
newQueues.add(newQueue.getQueueName());
}
}
- for (Iterator<Map.Entry<String, LeafQueueState>> itr =
- leafQueueStateMap.entrySet().iterator(); itr.hasNext(); ) {
- Map.Entry<String, LeafQueueState> e = itr.next();
- String queueName = e.getKey();
- if (!newQueues.contains(queueName)) {
+ for (Iterator<Map.Entry<String, Map<String, LeafQueueStatePerPartition>>>
+ itr = leafQueueState.getLeafQueueStateMap().entrySet().iterator();
+ itr.hasNext(); ) {
+ Map.Entry<String, Map<String, LeafQueueStatePerPartition>> e =
+ itr.next();
+ String partition = e.getKey();
+ if (!newPartitions.contains(partition)) {
itr.remove();
+ LOG.info(
+ "Removed partition " + partition + " from leaf queue " + "state");
+ } else{
+ Map<String, LeafQueueStatePerPartition> queues = e.getValue();
+ for (
+ Iterator<Map.Entry<String, LeafQueueStatePerPartition>> queueItr =
+ queues.entrySet().iterator(); queueItr.hasNext(); ) {
+ String queue = queueItr.next().getKey();
+ if (!newQueues.contains(queue)) {
+ queueItr.remove();
+ LOG.info("Removed queue " + queue + " from leaf queue "
+ + "state from partition " + partition);
+ }
+ }
}
}
} finally {
@@ -406,22 +513,20 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
}
}
- private LinkedHashSet<String> getSortedLeafQueues(
+ private LinkedHashSet<String> getSortedLeafQueues(String nodeLabel,
final List<FiCaSchedulerApp> pendingApps, int leafQueuesNeeded,
Set<String> deactivatedQueues) throws SchedulerDynamicEditException {
LinkedHashSet<String> leafQueues = new LinkedHashSet<>(leafQueuesNeeded);
int ctr = 0;
for (FiCaSchedulerApp app : pendingApps) {
-
AutoCreatedLeafQueue leafQueue =
(AutoCreatedLeafQueue) app.getCSLeafQueue();
String leafQueueName = leafQueue.getQueueName();
//Check if leafQueue is not active already and has any pending apps
if (ctr < leafQueuesNeeded) {
-
- if (!isActive(leafQueue)) {
+ if (!isActive(leafQueue, nodeLabel)) {
if (!deactivatedQueues.contains(leafQueueName)) {
if (addLeafQueueIfNotExists(leafQueues, leafQueueName)) {
ctr++;
@@ -445,11 +550,12 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
}
@VisibleForTesting
- public boolean isActive(final AutoCreatedLeafQueue leafQueue)
- throws SchedulerDynamicEditException {
+ public boolean isActive(final AutoCreatedLeafQueue leafQueue,
+ String nodeLabel) throws SchedulerDynamicEditException {
try {
readLock.lock();
- LeafQueueState leafQueueStatus = getLeafQueueState(leafQueue);
+ LeafQueueStatePerPartition leafQueueStatus = getLeafQueueState(leafQueue,
+ nodeLabel);
return leafQueueStatus.isActive();
} finally {
readLock.unlock();
@@ -457,64 +563,52 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
}
private Map<String, QueueCapacities> deactivateLeafQueuesIfInActive(
- ParentQueue parentQueue,
- List<QueueManagementChange> queueManagementChanges)
+ ParentQueue parentQueue, String nodeLabel,
+ Map<String, QueueCapacities> leafQueueEntitlements)
throws SchedulerDynamicEditException {
Map<String, QueueCapacities> deactivatedQueues = new HashMap<>();
for (CSQueue childQueue : parentQueue.getChildQueues()) {
AutoCreatedLeafQueue leafQueue = (AutoCreatedLeafQueue) childQueue;
+ if (leafQueue != null) {
+ if (isActive(leafQueue, nodeLabel) && !hasPendingApps(leafQueue)) {
+ if (!leafQueueEntitlements.containsKey(leafQueue.getQueueName())) {
+ leafQueueEntitlements.put(leafQueue.getQueueName(),
+ new QueueCapacities(false));
+ }
- if (isActive(leafQueue) && !hasPendingApps(leafQueue)) {
- queueManagementChanges.add(
- new QueueManagementChange.UpdateQueue(leafQueue,
- ZERO_CAPACITY_ENTITLEMENT));
- deactivatedQueues.put(leafQueue.getQueueName(),
- leafQueueTemplateCapacities);
- } else{
- if (LOG.isDebugEnabled()) {
- LOG.debug(" Leaf queue has pending applications : " + leafQueue
- .getNumApplications() + ".Skipping deactivation for "
- + leafQueue);
+ QueueCapacities capacities = leafQueueEntitlements.get(
+ leafQueue.getQueueName());
+ updateToZeroCapacity(capacities, nodeLabel);
+ deactivatedQueues.put(leafQueue.getQueueName(),
+ leafQueueTemplateCapacities);
+ } else{
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(" Leaf queue has pending applications or is " + "inactive"
+ + " : " + leafQueue.getNumApplications()
+ + ".Skipping deactivation for " + leafQueue);
+ }
}
+ } else{
+ LOG.warn("Could not find queue in scheduler while trying" + " to "
+ + "deactivate for " + parentQueue);
}
}
- if (LOG.isDebugEnabled()) {
- if (deactivatedQueues.size() > 0) {
- LOG.debug("Deactivated leaf queues : " + deactivatedQueues);
- }
- }
return deactivatedQueues;
}
- private void computeQueueManagementChanges(
+ private void updateLeafQueueCapacitiesByLabel(String nodeLabel,
Set<String> leafQueuesToBeActivated,
- List<QueueManagementChange> queueManagementChanges,
- final float availableCapacity,
- final float leafQueueTemplateAbsoluteCapacity) {
-
- float curAvailableCapacity = availableCapacity;
-
+ Map<String, QueueCapacities> leafQueueEntitlements) {
for (String curLeafQueue : leafQueuesToBeActivated) {
- // Activate queues if capacity is available
- if (curAvailableCapacity >= leafQueueTemplateAbsoluteCapacity) {
- AutoCreatedLeafQueue leafQueue =
- (AutoCreatedLeafQueue) scheduler.getCapacitySchedulerQueueManager()
- .getQueue(curLeafQueue);
- if (leafQueue != null) {
- AutoCreatedLeafQueueConfig newTemplate = buildTemplate(
- leafQueueTemplateCapacities.getCapacity(),
- leafQueueTemplateCapacities.getMaximumCapacity());
- queueManagementChanges.add(
- new QueueManagementChange.UpdateQueue(leafQueue, newTemplate));
- curAvailableCapacity -= leafQueueTemplateAbsoluteCapacity;
- } else{
- LOG.warn(
- "Could not find queue in scheduler while trying to deactivate "
- + curLeafQueue);
- }
+ if (!leafQueueEntitlements.containsKey(curLeafQueue)) {
+ leafQueueEntitlements.put(curLeafQueue, new QueueCapacities(false));
+ // Activate queues if capacity is available
}
+
+ QueueCapacities capacities = leafQueueEntitlements.get(curLeafQueue);
+ updateCapacityFromTemplate(capacities, nodeLabel);
}
}
@@ -528,17 +622,8 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
availableCapacity / childQueueAbsoluteCapacity);
return Math.min(numLeafQueuesNeeded, numPendingApps);
- } else{
- throw new SchedulerDynamicEditException("Child queue absolute capacity "
- + "is initialized to 0. Check parent queue's " + managedParentQueue
- .getQueueName() + " leaf queue template configuration");
}
- }
-
- private float getAvailableCapacity(float parentAbsCapacity,
- float deactivatedAbsCapacity, float totalChildQueueActivatedCapacity) {
- return parentAbsCapacity - totalChildQueueActivatedCapacity
- + deactivatedAbsCapacity + EPSILON;
+ return 0;
}
/**
@@ -567,25 +652,27 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
AutoCreatedLeafQueue leafQueue = (AutoCreatedLeafQueue) queue;
- if (updatedQueueTemplate.getQueueCapacities().getCapacity() > 0) {
- if (isActive(leafQueue)) {
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Queue is already active. Skipping activation : " + queue
- .getQueuePath());
+ for (String nodeLabel : updatedQueueTemplate.getQueueCapacities()
+ .getExistingNodeLabels()) {
+ if (updatedQueueTemplate.getQueueCapacities().
+ getCapacity(nodeLabel) > 0) {
+ if (isActive(leafQueue, nodeLabel)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Queue is already active." + " Skipping activation : "
+ + queue.getQueuePath());
+ }
+ } else{
+ activate(leafQueue, nodeLabel);
}
} else{
- activate(leafQueue);
- }
- } else{
- if (!isActive(leafQueue)) {
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Queue is already de-activated. " + "Skipping de-activation "
- + ": " + leafQueue.getQueuePath());
+ if (!isActive(leafQueue, nodeLabel)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Queue is already de-activated. Skipping "
+ + "de-activation : " + leafQueue.getQueuePath());
+ }
+ } else{
+ deactivate(leafQueue, nodeLabel);
}
- } else{
- deactivate(leafQueue);
}
}
}
@@ -594,30 +681,26 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
}
}
- private void activate(final AutoCreatedLeafQueue leafQueue)
- throws SchedulerDynamicEditException {
+ private void activate(final AbstractAutoCreatedLeafQueue leafQueue,
+ String nodeLabel) throws SchedulerDynamicEditException {
try {
writeLock.lock();
- getLeafQueueState(leafQueue).activate();
-
- parentQueueState.incAbsoluteActivatedChildCapacity(NO_LABEL,
- leafQueueTemplateCapacities.getAbsoluteCapacity());
+ getLeafQueueState(leafQueue, nodeLabel).activate();
+ parentQueueState.incAbsoluteActivatedChildCapacity(nodeLabel,
+ leafQueueTemplateCapacities.getAbsoluteCapacity(nodeLabel));
} finally {
writeLock.unlock();
}
}
- private void deactivate(final AutoCreatedLeafQueue leafQueue)
- throws SchedulerDynamicEditException {
+ private void deactivate(final AbstractAutoCreatedLeafQueue leafQueue,
+ String nodeLabel) throws SchedulerDynamicEditException {
try {
writeLock.lock();
- getLeafQueueState(leafQueue).deactivate();
+ getLeafQueueState(leafQueue, nodeLabel).deactivate();
- for (String nodeLabel : managedParentQueue.getQueueCapacities()
- .getExistingNodeLabels()) {
- parentQueueState.decAbsoluteActivatedChildCapacity(nodeLabel,
- leafQueueTemplateCapacities.getAbsoluteCapacity());
- }
+ parentQueueState.decAbsoluteActivatedChildCapacity(nodeLabel,
+ leafQueueTemplateCapacities.getAbsoluteCapacity(nodeLabel));
} finally {
writeLock.unlock();
}
@@ -629,7 +712,7 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
@Override
public void reinitialize(CapacitySchedulerContext schedulerContext,
- final ParentQueue parentQueue) {
+ final ParentQueue parentQueue) throws IOException {
if (!(parentQueue instanceof ManagedParentQueue)) {
throw new IllegalStateException(
"Expected instance of type " + ManagedParentQueue.class + " found "
@@ -649,12 +732,11 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
//clear state
parentQueueState.clear();
- clearLeafQueueState();
+ leafQueueState.clear();
LOG.info(
- "Reinitialized queue management policy for parent queue "
- + parentQueue.getQueueName() +" with leaf queue template "
- + "capacities : ["
+ "Reinitialized queue management policy for parent queue " + parentQueue
+ .getQueueName() + " with leaf queue template " + "capacities : ["
+ leafQueueTemplate.getQueueCapacities() + "]");
}
@@ -663,51 +745,74 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
AbstractAutoCreatedLeafQueue leafQueue)
throws SchedulerDynamicEditException {
- if ( !(leafQueue instanceof AutoCreatedLeafQueue)) {
- throw new SchedulerDynamicEditException("Not an instance of "
- + "AutoCreatedLeafQueue : " + leafQueue.getClass());
+ AutoCreatedLeafQueueConfig template;
+
+ if (!(leafQueue instanceof AutoCreatedLeafQueue)) {
+ throw new SchedulerDynamicEditException(
+ "Not an instance of " + "AutoCreatedLeafQueue : " + leafQueue
+ .getClass());
}
- AutoCreatedLeafQueue autoCreatedLeafQueue =
- (AutoCreatedLeafQueue) leafQueue;
- AutoCreatedLeafQueueConfig template = ZERO_CAPACITY_ENTITLEMENT;
try {
writeLock.lock();
- if (!addLeafQueueStateIfNotExists(leafQueue)) {
- LOG.error("Leaf queue already exists in state : " + getLeafQueueState(
- leafQueue));
- throw new SchedulerDynamicEditException(
- "Leaf queue already exists in state : " + getLeafQueueState(
- leafQueue));
- }
- float availableCapacity = getAvailableCapacity(
- managedParentQueue.getQueueCapacities().getAbsoluteCapacity(), 0,
- parentQueueState.getAbsoluteActivatedChildQueueCapacity());
+ QueueCapacities capacities = new QueueCapacities(false);
+ for (String nodeLabel : leafQueueTemplateNodeLabels) {
+ if (!leafQueueState.createLeafQueueStateIfNotExists(leafQueue,
+ nodeLabel)) {
+ String message =
+ "Leaf queue already exists in state : " + getLeafQueueState(
+ leafQueue, nodeLabel);
+ LOG.error(message);
+ }
- if (availableCapacity >= leafQueueTemplateCapacities
- .getAbsoluteCapacity()) {
- activate(autoCreatedLeafQueue);
- template = buildTemplate(leafQueueTemplateCapacities.getCapacity(),
- leafQueueTemplateCapacities.getMaximumCapacity());
+ float availableCapacity = managedParentQueue.getQueueCapacities().
+ getAbsoluteCapacity(nodeLabel) - parentQueueState.
+ getAbsoluteActivatedChildQueueCapacity(nodeLabel) + EPSILON;
+
+ if (availableCapacity >= leafQueueTemplateCapacities
+ .getAbsoluteCapacity(nodeLabel)) {
+ updateCapacityFromTemplate(capacities, nodeLabel);
+ activate(leafQueue, nodeLabel);
+ } else{
+ updateToZeroCapacity(capacities, nodeLabel);
+ }
}
+
+ template = buildTemplate(capacities);
} finally {
writeLock.unlock();
}
return template;
}
+ private void updateToZeroCapacity(QueueCapacities capacities,
+ String nodeLabel) {
+ capacities.setCapacity(nodeLabel, 0.0f);
+ capacities.setMaximumCapacity(nodeLabel,
+ leafQueueTemplateCapacities.getMaximumCapacity(nodeLabel));
+ }
+
+ private void updateCapacityFromTemplate(QueueCapacities capacities,
+ String nodeLabel) {
+ capacities.setCapacity(nodeLabel,
+ leafQueueTemplateCapacities.getCapacity(nodeLabel));
+ capacities.setMaximumCapacity(nodeLabel,
+ leafQueueTemplateCapacities.getMaximumCapacity(nodeLabel));
+ }
+
@VisibleForTesting
- LeafQueueState getLeafQueueState(LeafQueue queue)
- throws SchedulerDynamicEditException {
+ LeafQueueStatePerPartition getLeafQueueState(LeafQueue queue,
+ String partition) throws SchedulerDynamicEditException {
try {
readLock.lock();
String queueName = queue.getQueueName();
- if (!containsLeafQueue(queueName)) {
+ if (!leafQueueState.containsLeafQueue(queueName, partition)) {
throw new SchedulerDynamicEditException(
"Could not find leaf queue in " + "state " + queueName);
} else{
- return leafQueueStateMap.get(queueName);
+ return leafQueueState.
+ getLeafQueueStatePerPartition(queueName, partition);
}
} finally {
readLock.unlock();
@@ -715,8 +820,8 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
}
@VisibleForTesting
- public float getAbsoluteActivatedChildQueueCapacity() {
- return parentQueueState.getAbsoluteActivatedChildQueueCapacity();
+ public float getAbsoluteActivatedChildQueueCapacity(String nodeLabel) {
+ return parentQueueState.getAbsoluteActivatedChildQueueCapacity(nodeLabel);
}
private List<FiCaSchedulerApp> getSortedPendingApplications() {
@@ -726,20 +831,10 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
return apps;
}
- private AutoCreatedLeafQueueConfig buildTemplate(float capacity,
- float maxCapacity) {
+ private AutoCreatedLeafQueueConfig buildTemplate(QueueCapacities capacities) {
AutoCreatedLeafQueueConfig.Builder templateBuilder =
new AutoCreatedLeafQueueConfig.Builder();
-
- QueueCapacities capacities = new QueueCapacities(false);
templateBuilder.capacities(capacities);
-
- for (String nodeLabel : managedParentQueue.getQueueCapacities()
- .getExistingNodeLabels()) {
- capacities.setCapacity(nodeLabel, capacity);
- capacities.setMaximumCapacity(nodeLabel, maxCapacity);
- }
-
return new AutoCreatedLeafQueueConfig(templateBuilder);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/PendingAskUpdateResult.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/placement/PendingAskUpdateResult.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/PendingAskUpdateResult.java
index 8765e86..8702c03 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/PendingAskUpdateResult.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/PendingAskUpdateResult.java
@@ -62,4 +62,12 @@ public class PendingAskUpdateResult {
public String getNewNodePartition() {
return newNodePartition;
}
+
+ @Override
+ public String toString() {
+ return "PendingAskUpdateResult{" + "lastPendingAsk=" + lastPendingAsk
+ + ", lastNodePartition='" + lastNodePartition + '\''
+ + ", newPendingAsk=" + newPendingAsk + ", newNodePartition='"
+ + newNodePartition + '\'' + '}';
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.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/MockNM.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java
index 0a06e82..2e28395 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java
@@ -24,6 +24,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
@@ -33,6 +34,7 @@ import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerState;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.NodeLabel;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.api.protocolrecords.NMContainerStatus;
@@ -65,6 +67,7 @@ public class MockNM {
new HashMap<ContainerId, ContainerStatus>();
private Map<ApplicationId, AppCollectorData> registeringCollectors
= new ConcurrentHashMap<>();
+ private Set<NodeLabel> nodeLabels;
public MockNM(String nodeIdStr, int memory, ResourceTrackerService resourceTracker) {
// scale vcores based on the requested memory
@@ -101,6 +104,13 @@ public class MockNM {
nodeId = BuilderUtils.newNodeId(splits[0], Integer.parseInt(splits[1]));
}
+ public MockNM(String nodeIdStr, Resource capability,
+ ResourceTrackerService resourceTracker, String version, Set<NodeLabel>
+ nodeLabels) {
+ this(nodeIdStr, capability, resourceTracker, version);
+ this.nodeLabels = nodeLabels;
+ }
+
public NodeId getNodeId() {
return nodeId;
}
@@ -164,12 +174,17 @@ public class MockNM {
List<ApplicationId> runningApplications) throws Exception {
RegisterNodeManagerRequest req = Records.newRecord(
RegisterNodeManagerRequest.class);
+
req.setNodeId(nodeId);
req.setHttpPort(httpPort);
req.setResource(capability);
req.setContainerStatuses(containerReports);
req.setNMVersion(version);
req.setRunningApplications(runningApplications);
+ if ( nodeLabels != null && nodeLabels.size() > 0) {
+ req.setNodeLabels(nodeLabels);
+ }
+
RegisterNodeManagerResponse registrationResponse =
resourceTracker.registerNodeManager(req);
this.currentContainerTokenMasterKey =
http://git-wip-us.apache.org/repos/asf/hadoop/blob/821b0de4/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestAppManager.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/TestAppManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestAppManager.java
index 1a1b527..f6cdfec 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestAppManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestAppManager.java
@@ -36,6 +36,7 @@ import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
@@ -57,6 +58,7 @@ import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.ExecutionType;
import org.apache.hadoop.yarn.api.records.ExecutionTypeRequest;
import org.apache.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.QueueInfo;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceInformation;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
@@ -247,10 +249,11 @@ public class TestAppManager{
private TestRMAppManager appMonitor;
private ApplicationSubmissionContext asContext;
private ApplicationId appId;
+ private QueueInfo mockDefaultQueueInfo;
@SuppressWarnings("deprecation")
@Before
- public void setUp() {
+ public void setUp() throws IOException {
long now = System.currentTimeMillis();
rmContext = mockRMContext(1, now - 10);
@@ -258,6 +261,7 @@ public class TestAppManager{
.setRMTimelineCollectorManager(mock(RMTimelineCollectorManager.class));
ResourceScheduler scheduler = mockResourceScheduler();
((RMContextImpl)rmContext).setScheduler(scheduler);
+
Configuration conf = new Configuration();
conf.setBoolean(YarnConfiguration.NODE_LABELS_ENABLED, true);
((RMContextImpl) rmContext).setYarnConfiguration(conf);
@@ -275,6 +279,11 @@ public class TestAppManager{
asContext.setAMContainerSpec(mockContainerLaunchContext(recordFactory));
asContext.setResource(mockResource());
asContext.setPriority(Priority.newInstance(0));
+ asContext.setQueue("default");
+ mockDefaultQueueInfo = mock(QueueInfo.class);
+ when(scheduler.getQueueInfo("default", false, false))
+ .thenReturn(mockDefaultQueueInfo);
+
setupDispatcher(rmContext, conf);
}
@@ -709,6 +718,7 @@ public class TestAppManager{
for (ResourceRequest req : reqs) {
req.setNodeLabelExpression(RMNodeLabelsManager.NO_LABEL);
}
+
// setAMContainerResourceRequests has priority over
// setAMContainerResourceRequest and setResource
Assert.assertEquals(reqs, app.getAMResourceRequests());
@@ -722,6 +732,7 @@ public class TestAppManager{
ResourceRequest req =
ResourceRequest.newInstance(Priority.newInstance(0),
ResourceRequest.ANY, Resources.createResource(1025), 1, true);
+ req.setNodeLabelExpression(RMNodeLabelsManager.NO_LABEL);
asContext.setAMContainerResourceRequest(cloneResourceRequest(req));
// getAMContainerResourceRequests uses a singleton list of
// getAMContainerResourceRequest
@@ -729,7 +740,6 @@ public class TestAppManager{
Assert.assertEquals(req, asContext.getAMContainerResourceRequests().get(0));
Assert.assertEquals(1, asContext.getAMContainerResourceRequests().size());
RMApp app = testRMAppSubmit();
- req.setNodeLabelExpression(RMNodeLabelsManager.NO_LABEL);
// setAMContainerResourceRequest has priority over setResource
Assert.assertEquals(Collections.singletonList(req),
app.getAMResourceRequests());
@@ -740,10 +750,12 @@ public class TestAppManager{
asContext.setResource(Resources.createResource(1024));
asContext.setAMContainerResourceRequests(null);
RMApp app = testRMAppSubmit();
+
// setResource
Assert.assertEquals(Collections.singletonList(
ResourceRequest.newInstance(RMAppAttemptImpl.AM_CONTAINER_PRIORITY,
- ResourceRequest.ANY, Resources.createResource(1024), 1, true, "")),
+ ResourceRequest.ANY, Resources.createResource(1024), 1, true,
+ "")),
app.getAMResourceRequests());
}
@@ -766,6 +778,8 @@ public class TestAppManager{
throws Exception {
asContext.setResource(null);
List<ResourceRequest> reqs = new ArrayList<>();
+ when(mockDefaultQueueInfo.getAccessibleNodeLabels()).thenReturn
+ (new HashSet<String>() {{ add("label1"); add(""); }});
ResourceRequest anyReq = ResourceRequest.newInstance(
Priority.newInstance(1),
ResourceRequest.ANY, Resources.createResource(1024), 1, false, "label1",
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org