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 ka...@apache.org on 2015/01/06 00:11:22 UTC
hadoop git commit: YARN-2881. [YARN-2574] Implement PlanFollower for
FairScheduler. (Anubhav Dhoot via kasha)
Repository: hadoop
Updated Branches:
refs/heads/trunk 53caeaa16 -> 0c4b11267
YARN-2881. [YARN-2574] Implement PlanFollower for FairScheduler. (Anubhav Dhoot via kasha)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/0c4b1126
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/0c4b1126
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/0c4b1126
Branch: refs/heads/trunk
Commit: 0c4b11267717eb451fa6ed4c586317f2db32fbd5
Parents: 53caeaa
Author: Karthik Kambatla <ka...@apache.org>
Authored: Tue Jan 6 04:41:45 2015 +0530
Committer: Karthik Kambatla <ka...@apache.org>
Committed: Tue Jan 6 04:42:10 2015 +0530
----------------------------------------------------------------------
hadoop-yarn-project/CHANGES.txt | 2 +
.../reservation/AbstractReservationSystem.java | 2 +
.../AbstractSchedulerPlanFollower.java | 3 +-
.../reservation/FairSchedulerPlanFollower.java | 141 +++++++++++++
.../reservation/ReservationConstants.java | 28 +++
.../scheduler/capacity/CapacityScheduler.java | 6 +-
.../scheduler/capacity/PlanQueue.java | 2 -
.../scheduler/fair/AllocationConfiguration.java | 14 ++
.../scheduler/fair/FSLeafQueue.java | 10 +
.../scheduler/fair/FairScheduler.java | 104 +++++++++-
.../scheduler/fair/QueueManager.java | 13 +-
.../fair/ReservationQueueConfiguration.java | 11 +
.../reservation/ReservationSystemTestUtil.java | 26 ++-
.../TestCapacitySchedulerPlanFollower.java | 3 +-
.../reservation/TestFairReservationSystem.java | 55 ++---
.../TestFairSchedulerPlanFollower.java | 203 +++++++++++++++++++
.../TestCapacitySchedulerDynamicBehavior.java | 3 +-
.../scheduler/fair/FairSchedulerTestBase.java | 4 +-
.../fair/TestFairSchedulerPreemption.java | 3 +-
19 files changed, 574 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index 63faa60..12c8066 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -70,6 +70,8 @@ Release 2.7.0 - UNRELEASED
YARN-2738. [YARN-2574] Add FairReservationSystem for FairScheduler.
(Anubhav Dhoot via kasha)
+ YARN-2881. [YARN-2574] Implement PlanFollower for FairScheduler.
+ (Anubhav Dhoot via kasha)
IMPROVEMENTS
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/AbstractReservationSystem.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/reservation/AbstractReservationSystem.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/AbstractReservationSystem.java
index fa0835a..8a15ac6 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/AbstractReservationSystem.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/AbstractReservationSystem.java
@@ -204,6 +204,8 @@ public abstract class AbstractReservationSystem extends AbstractService
// currently only capacity scheduler is supported
if (scheduler instanceof CapacityScheduler) {
return CapacitySchedulerPlanFollower.class.getName();
+ } else if (scheduler instanceof FairScheduler) {
+ return FairSchedulerPlanFollower.class.getName();
}
return null;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/AbstractSchedulerPlanFollower.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/reservation/AbstractSchedulerPlanFollower.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/AbstractSchedulerPlanFollower.java
index 0de4dcf..ea7f27d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/AbstractSchedulerPlanFollower.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/AbstractSchedulerPlanFollower.java
@@ -26,7 +26,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.Plan
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Queue;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.PlanQueue;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.QueueEntitlement;
import org.apache.hadoop.yarn.util.Clock;
@@ -99,7 +98,7 @@ public abstract class AbstractSchedulerPlanFollower implements PlanFollower {
// create the default reservation queue if it doesnt exist
String defReservationId = getReservationIdFromQueueName(planQueueName) +
- PlanQueue.DEFAULT_QUEUE_SUFFIX;
+ ReservationConstants.DEFAULT_QUEUE_SUFFIX;
String defReservationQueue = getReservationQueueName(planQueueName,
defReservationId);
createDefaultReservationQueue(planQueueName, planQueue,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/FairSchedulerPlanFollower.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/reservation/FairSchedulerPlanFollower.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/FairSchedulerPlanFollower.java
new file mode 100644
index 0000000..7ca03c5
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/FairSchedulerPlanFollower.java
@@ -0,0 +1,141 @@
+/**
+ * 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.reservation;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.hadoop.yarn.api.records.ReservationId;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Queue;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSLeafQueue;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSParentQueue;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSQueue;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
+import org.apache.hadoop.yarn.util.Clock;
+import org.apache.hadoop.yarn.util.resource.Resources;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FairSchedulerPlanFollower extends AbstractSchedulerPlanFollower {
+ private static final Logger LOG = LoggerFactory
+ .getLogger(FairSchedulerPlanFollower.class);
+
+ private FairScheduler fs;
+
+ @Override
+ public void init(Clock clock, ResourceScheduler sched,
+ Collection<Plan> plans) {
+ super.init(clock, sched, plans);
+ fs = (FairScheduler)sched;
+ LOG.info("Initializing Plan Follower Policy:"
+ + this.getClass().getCanonicalName());
+ }
+
+ @Override
+ protected Queue getPlanQueue(String planQueueName) {
+ Queue planQueue = fs.getQueueManager().getParentQueue(planQueueName, false);
+ if (planQueue == null) {
+ LOG.error("The queue " + planQueueName + " cannot be found or is not a " +
+ "ParentQueue");
+ }
+ return planQueue;
+ }
+
+ @Override
+ protected float calculateReservationToPlanRatio(Resource clusterResources,
+ Resource planResources, Resource capToAssign) {
+ return Resources.divide(fs.getResourceCalculator(),
+ clusterResources, capToAssign, planResources);
+ }
+
+ @Override
+ protected boolean arePlanResourcesLessThanReservations(Resource
+ clusterResources, Resource planResources, Resource reservedResources) {
+ return Resources.greaterThan(fs.getResourceCalculator(),
+ clusterResources, reservedResources, planResources);
+ }
+
+ @Override
+ protected List<? extends Queue> getChildReservationQueues(Queue queue) {
+ FSQueue planQueue = (FSQueue)queue;
+ List<FSQueue> childQueues = planQueue.getChildQueues();
+ return childQueues;
+ }
+
+
+ @Override
+ protected void addReservationQueue(String planQueueName, Queue queue,
+ String currResId) {
+ String leafQueueName = getReservationQueueName(planQueueName, currResId);
+ fs.getQueueManager().getLeafQueue(leafQueueName, true);
+ }
+
+ @Override
+ protected void createDefaultReservationQueue(String planQueueName,
+ Queue queue, String defReservationId) {
+ String defReservationQueueName = getReservationQueueName(planQueueName,
+ defReservationId);
+ if (!fs.getQueueManager().exists(defReservationQueueName)) {
+ fs.getQueueManager().getLeafQueue(defReservationQueueName, true);
+ }
+ }
+
+ @Override
+ protected Resource getPlanResources(Plan plan, Queue queue,
+ Resource clusterResources) {
+ FSParentQueue planQueue = (FSParentQueue)queue;
+ Resource planResources = planQueue.getSteadyFairShare();
+ return planResources;
+ }
+
+ @Override
+ protected Resource getReservationQueueResourceIfExists(Plan plan,
+ ReservationId reservationId) {
+ String reservationQueueName = getReservationQueueName(plan.getQueueName(),
+ reservationId.toString());
+ FSLeafQueue reservationQueue =
+ fs.getQueueManager().getLeafQueue(reservationQueueName, false);
+ Resource reservationResource = null;
+ if (reservationQueue != null) {
+ reservationResource = reservationQueue.getSteadyFairShare();
+ }
+ return reservationResource;
+ }
+
+ @Override
+ protected String getReservationQueueName(String planQueueName,
+ String reservationQueueName) {
+ String planQueueNameFullPath = fs.getQueueManager().getQueue
+ (planQueueName).getName();
+
+ if (!reservationQueueName.startsWith(planQueueNameFullPath)) {
+ // If name is not a path we need full path for FairScheduler. See
+ // YARN-2773 for the root cause
+ return planQueueNameFullPath + "." + reservationQueueName;
+ }
+ return reservationQueueName;
+ }
+
+ @Override
+ protected String getReservationIdFromQueueName(String resQueueName) {
+ return resQueueName.substring(resQueueName.lastIndexOf(".") + 1);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationConstants.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/reservation/ReservationConstants.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationConstants.java
new file mode 100644
index 0000000..1ff941f
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationConstants.java
@@ -0,0 +1,28 @@
+/*
+ * 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.reservation;
+
+public interface ReservationConstants {
+
+ /**
+ * The suffix used for a queue under a reservable queue that will be used
+ * as a default queue whenever no reservation is used
+ */
+ String DEFAULT_QUEUE_SUFFIX = "-default";
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/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 3648c54..5e6a352 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
@@ -21,7 +21,6 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
import java.util.EnumSet;
@@ -66,6 +65,7 @@ import org.apache.hadoop.yarn.proto.YarnServiceProtos.SchedulerResourceTypes;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
+import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationConstants;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType;
@@ -1419,7 +1419,7 @@ public class CapacityScheduler extends
queueName = resQName;
} else {
// use the default child queue of the plan for unreserved apps
- queueName = queueName + PlanQueue.DEFAULT_QUEUE_SUFFIX;
+ queueName = queueName + ReservationConstants.DEFAULT_QUEUE_SUFFIX;
}
return queueName;
}
@@ -1583,7 +1583,7 @@ public class CapacityScheduler extends
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;
+ targetQueueName = targetQueueName + ReservationConstants.DEFAULT_QUEUE_SUFFIX;
}
return targetQueueName;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/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
index 0725959..f8b11eb 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/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
@@ -37,8 +37,6 @@ import org.slf4j.LoggerFactory;
*/
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;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationConfiguration.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationConfiguration.java
index fd99d65..0ea7314 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationConfiguration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/AllocationConfiguration.java
@@ -207,6 +207,10 @@ public class AllocationConfiguration extends ReservationSchedulerConfiguration {
ResourceWeights weight = queueWeights.get(queue);
return (weight == null) ? ResourceWeights.NEUTRAL : weight;
}
+
+ public void setQueueWeight(String queue, ResourceWeights weight) {
+ queueWeights.put(queue, weight);
+ }
public int getUserMaxApps(String user) {
Integer maxApps = userMaxApps.get(user);
@@ -323,4 +327,14 @@ public class AllocationConfiguration extends ReservationSchedulerConfiguration {
public long getEnforcementWindow(String queue) {
return globalReservationQueueConfig.getEnforcementWindowMsec();
}
+
+ @VisibleForTesting
+ public void setReservationWindow(long window) {
+ globalReservationQueueConfig.setReservationWindow(window);
+ }
+
+ @VisibleForTesting
+ public void setAverageCapacity(int avgCapacity) {
+ globalReservationQueueConfig.setAverageCapacity(avgCapacity);
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java
index 91bea11..3c97535 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java
@@ -37,6 +37,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.QueueACL;
import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.server.resourcemanager.resource.ResourceWeights;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerAppUtils;
@@ -516,6 +517,15 @@ public class FSLeafQueue extends FSQueue {
}
}
+ /** Allows setting weight for a dynamically created queue
+ * Currently only used for reservation based queues
+ * @param weight queue weight
+ */
+ public void setWeights(float weight) {
+ scheduler.getAllocationConfiguration().setQueueWeight(getName(),
+ new ResourceWeights(weight));
+ }
+
/**
* Helper method to check if the queue should preempt containers
*
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
index 57b41af..124da99 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
@@ -38,6 +38,7 @@ import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.QueueACL;
import org.apache.hadoop.yarn.api.records.QueueInfo;
import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
+import org.apache.hadoop.yarn.api.records.ReservationId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceOption;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
@@ -47,6 +48,7 @@ import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.apache.hadoop.yarn.proto.YarnServiceProtos.SchedulerResourceTypes;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
+import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationConstants;
import org.apache.hadoop.yarn.server.resourcemanager.resource.ResourceWeights;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
@@ -68,6 +70,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplication;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt.ContainersAndNMTokensAllocation;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.QueueEntitlement;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppAddedSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppAttemptAddedSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppAttemptRemovedSchedulerEvent;
@@ -1163,9 +1166,15 @@ public class FairScheduler extends
throw new RuntimeException("Unexpected event type: " + event);
}
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:
if (!(event instanceof AppRemovedSchedulerEvent)) {
@@ -1223,6 +1232,51 @@ public class FairScheduler extends
}
}
+ private String resolveReservationQueueName(String queueName,
+ ApplicationId applicationId, ReservationId reservationID) {
+ FSQueue queue = queueMgr.getQueue(queueName);
+ if ((queue == null) || !allocConf.isReservable(queue.getQueueName())) {
+ return queueName;
+ }
+ // Use fully specified name from now on (including root. prefix)
+ queueName = queue.getQueueName();
+ if (reservationID != null) {
+ String resQName = queueName + "." + reservationID.toString();
+ queue = queueMgr.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;
+ }
+ if (!queue.getParent().getQueueName().equals(queueName)) {
+ String message =
+ "Application: " + applicationId + " submitted to a reservation "
+ + resQName + " which does not belong to the specified queue: "
+ + queueName;
+ 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 = getDefaultQueueForPlanQueue(queueName);
+ }
+ return queueName;
+ }
+
+ private String getDefaultQueueForPlanQueue(String queueName) {
+ String planName = queueName.substring(queueName.lastIndexOf(".") + 1);
+ queueName = queueName + "." + planName + ReservationConstants.DEFAULT_QUEUE_SUFFIX;
+ return queueName;
+ }
+
@Override
public void recover(RMState state) throws Exception {
// NOT IMPLEMENTED
@@ -1441,7 +1495,8 @@ public class FairScheduler extends
// To serialize with FairScheduler#allocate, synchronize on app attempt
synchronized (attempt) {
FSLeafQueue oldQueue = (FSLeafQueue) app.getQueue();
- FSLeafQueue targetQueue = queueMgr.getLeafQueue(queueName, false);
+ String destQueueName = handleMoveToPlanQueue(queueName);
+ FSLeafQueue targetQueue = queueMgr.getLeafQueue(destQueueName, false);
if (targetQueue == null) {
throw new YarnException("Target queue " + queueName
+ " not found or is not a leaf queue.");
@@ -1577,4 +1632,45 @@ public class FairScheduler extends
}
return planQueues;
}
+
+ @Override
+ public void setEntitlement(String queueName,
+ QueueEntitlement entitlement) throws YarnException {
+
+ FSLeafQueue reservationQueue = queueMgr.getLeafQueue(queueName, false);
+ if (reservationQueue == null) {
+ throw new YarnException("Target queue " + queueName
+ + " not found or is not a leaf queue.");
+ }
+
+ reservationQueue.setWeights(entitlement.getCapacity());
+
+ // TODO Does MaxCapacity need to be set for fairScheduler ?
+ }
+
+ /**
+ * Only supports removing empty leaf queues
+ * @param queueName name of queue to remove
+ * @throws YarnException if queue to remove is either not a leaf or if its
+ * not empty
+ */
+ @Override
+ public void removeQueue(String queueName) throws YarnException {
+ FSLeafQueue reservationQueue = queueMgr.getLeafQueue(queueName, false);
+ if (reservationQueue != null) {
+ if (!queueMgr.removeLeafQueue(queueName)) {
+ throw new YarnException("Could not remove queue " + queueName + " as " +
+ "its either not a leaf queue or its not empty");
+ }
+ }
+ }
+
+ private String handleMoveToPlanQueue(String targetQueueName) {
+ FSQueue dest = queueMgr.getQueue(targetQueueName);
+ if (dest != null && allocConf.isReservable(dest.getQueueName())) {
+ // use the default child reservation queue of the plan
+ targetQueueName = getDefaultQueueForPlanQueue(targetQueueName);
+ }
+ return targetQueueName;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/QueueManager.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/QueueManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/QueueManager.java
index 0e625d7..27e571e 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/QueueManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/QueueManager.java
@@ -91,7 +91,18 @@ public class QueueManager {
}
return (FSLeafQueue) queue;
}
-
+
+ /**
+ * Remove a leaf queue if empty
+ * @param name name of the queue
+ * @return true if queue was removed or false otherwise
+ */
+ public boolean removeLeafQueue(String name) {
+ name = ensureRootPrefix(name);
+ return removeEmptyIncompatibleQueues(name, FSQueueType.PARENT);
+ }
+
+
/**
* Get a parent queue by name, creating it if the create param is true and is necessary.
* If the queue is not or can not be a parent queue, i.e. it already exists as a
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/ReservationQueueConfiguration.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/ReservationQueueConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/ReservationQueueConfiguration.java
index 747a4c2..cf7f84e 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/ReservationQueueConfiguration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/ReservationQueueConfiguration.java
@@ -17,6 +17,7 @@
*/
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSchedulerConfiguration;
@@ -102,4 +103,14 @@ public class ReservationQueueConfiguration {
public void setReservationAgent(String reservationAgent) {
this.reservationAgent = reservationAgent;
}
+
+ @VisibleForTesting
+ public void setReservationWindow(long reservationWindow) {
+ this.reservationWindow = reservationWindow;
+ }
+
+ @VisibleForTesting
+ public void setAverageCapacity(int averageCapacity) {
+ this.avgOverTimeMultiplier = averageCapacity;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationSystemTestUtil.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/reservation/ReservationSystemTestUtil.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationSystemTestUtil.java
index 12c2583..bfaf06b 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationSystemTestUtil.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationSystemTestUtil.java
@@ -29,7 +29,6 @@ import java.io.PrintWriter;
import java.util.Collections;
import java.util.Map;
import java.util.Random;
-import java.util.Set;
import java.util.TreeMap;
import org.apache.hadoop.conf.Configuration;
@@ -42,12 +41,16 @@ import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.impl.pb.ReservationDefinitionPBImpl;
import org.apache.hadoop.yarn.api.records.impl.pb.ReservationRequestsPBImpl;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.MockNodes;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.RMContextImpl;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
+import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeAddedSchedulerEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.security.ClientToAMTokenSecretManagerInRM;
import org.apache.hadoop.yarn.server.resourcemanager.security.NMTokenSecretManagerInRM;
import org.apache.hadoop.yarn.server.resourcemanager.security.RMContainerTokenSecretManager;
@@ -160,6 +163,27 @@ public class ReservationSystemTestUtil {
out.close();
}
+ public static FairScheduler setupFairScheduler(
+ ReservationSystemTestUtil testUtil,
+ RMContext rmContext, Configuration conf, int numContainers) throws
+ IOException {
+ FairScheduler scheduler = new FairScheduler();
+ scheduler.setRMContext(rmContext);
+
+ when(rmContext.getScheduler()).thenReturn(scheduler);
+
+ scheduler.init(conf);
+ scheduler.start();
+ scheduler.reinitialize(conf, rmContext);
+
+
+ Resource resource = testUtil.calculateClusterResource(numContainers);
+ RMNode node1 = MockNodes.newNodeInfo(1, resource, 1, "127.0.0.1");
+ NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1);
+ scheduler.handle(nodeEvent1);
+ return scheduler;
+ }
+
@SuppressWarnings("unchecked")
public CapacityScheduler mockCapacityScheduler(int numContainers)
throws IOException {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestCapacitySchedulerPlanFollower.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/reservation/TestCapacitySchedulerPlanFollower.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestCapacitySchedulerPlanFollower.java
index c603f5b..c7513ab 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestCapacitySchedulerPlanFollower.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestCapacitySchedulerPlanFollower.java
@@ -41,7 +41,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueue;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerContext;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.PlanQueue;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.TestUtils;
import org.apache.hadoop.yarn.server.resourcemanager.security.RMContainerTokenSecretManager;
import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
@@ -149,7 +148,7 @@ public class TestCapacitySchedulerPlanFollower extends TestSchedulerPlanFollower
@Override
protected Queue getDefaultQueue() {
- return cs.getQueue("dedicated" + PlanQueue.DEFAULT_QUEUE_SUFFIX);
+ return cs.getQueue("dedicated" + ReservationConstants.DEFAULT_QUEUE_SUFFIX);
}
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestFairReservationSystem.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/reservation/TestFairReservationSystem.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestFairReservationSystem.java
index 82ba731..f294eaf 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestFairReservationSystem.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestFairReservationSystem.java
@@ -18,14 +18,11 @@
package org.apache.hadoop.yarn.server.resourcemanager.reservation;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.server.resourcemanager.MockNodes;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
-import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeAddedSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairSchedulerConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairSchedulerTestBase;
@@ -38,15 +35,16 @@ import org.junit.Test;
import java.io.File;
import java.io.IOException;
-import static org.mockito.Mockito.when;
-
-public class TestFairReservationSystem extends FairSchedulerTestBase {
- private final static String ALLOC_FILE = new File(TEST_DIR,
+public class TestFairReservationSystem {
+ private final static String ALLOC_FILE = new File(FairSchedulerTestBase.
+ TEST_DIR,
TestFairReservationSystem.class.getName() + ".xml").getAbsolutePath();
+ private Configuration conf;
+ private FairScheduler scheduler;
+ private FairSchedulerTestBase testHelper = new FairSchedulerTestBase();
- @Override
protected Configuration createConfiguration() {
- Configuration conf = super.createConfiguration();
+ Configuration conf = testHelper.createConfiguration();
conf.setClass(YarnConfiguration.RM_SCHEDULER, FairScheduler.class,
ResourceScheduler.class);
conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, ALLOC_FILE);
@@ -60,10 +58,6 @@ public class TestFairReservationSystem extends FairSchedulerTestBase {
@After
public void teardown() {
- if (resourceManager != null) {
- resourceManager.stop();
- resourceManager = null;
- }
conf = null;
}
@@ -75,7 +69,8 @@ public class TestFairReservationSystem extends FairSchedulerTestBase {
// Setup
RMContext mockRMContext = testUtil.createRMContext(conf);
- setupFairScheduler(testUtil, mockRMContext);
+ scheduler = ReservationSystemTestUtil.setupFairScheduler(testUtil,
+ mockRMContext, conf, 10);
FairReservationSystem reservationSystem = new FairReservationSystem();
reservationSystem.setRMContext(mockRMContext);
@@ -97,14 +92,15 @@ public class TestFairReservationSystem extends FairSchedulerTestBase {
ReservationSystemTestUtil testUtil = new ReservationSystemTestUtil();
// Setup
- RMContext mockContext = testUtil.createRMContext(conf);
- setupFairScheduler(testUtil, mockContext);
+ RMContext mockRMContext = testUtil.createRMContext(conf);
+ scheduler = ReservationSystemTestUtil.setupFairScheduler(testUtil,
+ mockRMContext, conf, 10);
FairReservationSystem reservationSystem = new FairReservationSystem();
- reservationSystem.setRMContext(mockContext);
+ reservationSystem.setRMContext(mockRMContext);
try {
- reservationSystem.reinitialize(scheduler.getConf(), mockContext);
+ reservationSystem.reinitialize(scheduler.getConf(), mockRMContext);
} catch (YarnException e) {
Assert.fail(e.getMessage());
}
@@ -116,10 +112,10 @@ public class TestFairReservationSystem extends FairSchedulerTestBase {
// Dynamically add a plan
ReservationSystemTestUtil.updateFSAllocationFile(ALLOC_FILE);
- scheduler.reinitialize(conf, mockContext);
+ scheduler.reinitialize(conf, mockRMContext);
try {
- reservationSystem.reinitialize(conf, mockContext);
+ reservationSystem.reinitialize(conf, mockRMContext);
} catch (YarnException e) {
Assert.fail(e.getMessage());
}
@@ -129,23 +125,4 @@ public class TestFairReservationSystem extends FairSchedulerTestBase {
(reservationSystem, newQueue);
}
- private void setupFairScheduler(ReservationSystemTestUtil testUtil,
- RMContext rmContext) throws
- IOException {
-
- scheduler = new FairScheduler();
- scheduler.setRMContext(rmContext);
-
- int numContainers = 10;
- when(rmContext.getScheduler()).thenReturn(scheduler);
-
- scheduler.init(conf);
- scheduler.start();
- scheduler.reinitialize(conf, rmContext);
-
- Resource resource = testUtil.calculateClusterResource(numContainers);
- RMNode node1 = MockNodes.newNodeInfo(1, resource, 1, "127.0.0.1");
- NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1);
- scheduler.handle(nodeEvent1);
- }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestFairSchedulerPlanFollower.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/reservation/TestFairSchedulerPlanFollower.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestFairSchedulerPlanFollower.java
new file mode 100644
index 0000000..e9a4f50
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestFairSchedulerPlanFollower.java
@@ -0,0 +1,203 @@
+/**
+* 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.reservation;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ReservationId;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.PlanningException;
+import org.apache.hadoop.yarn.server.resourcemanager.resource.ResourceType;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Queue;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.TestUtils;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.AllocationConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSLeafQueue;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSQueue;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairSchedulerConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairSchedulerTestBase;
+import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
+import org.apache.hadoop.yarn.util.Clock;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
+import org.mockito.Matchers;
+import org.mockito.Mockito;
+
+public class TestFairSchedulerPlanFollower extends
+ TestSchedulerPlanFollowerBase {
+ private final static String ALLOC_FILE = new File(FairSchedulerTestBase.
+ TEST_DIR,
+ TestFairReservationSystem.class.getName() + ".xml").getAbsolutePath();
+ private RMContext rmContext;
+ private RMContext spyRMContext;
+ private FairScheduler fs;
+ private Configuration conf;
+ private FairSchedulerTestBase testHelper = new FairSchedulerTestBase();
+
+ @Rule
+ public TestName name = new TestName();
+
+ protected Configuration createConfiguration() {
+ Configuration conf = testHelper.createConfiguration();
+ conf.setClass(YarnConfiguration.RM_SCHEDULER, FairScheduler.class,
+ ResourceScheduler.class);
+ conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, ALLOC_FILE);
+ return conf;
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ conf = createConfiguration();
+ ReservationSystemTestUtil.setupFSAllocationFile(ALLOC_FILE);
+ ReservationSystemTestUtil testUtil = new ReservationSystemTestUtil();
+
+ // Setup
+ rmContext = TestUtils.getMockRMContext();
+ spyRMContext = spy(rmContext);
+ fs = ReservationSystemTestUtil.setupFairScheduler(testUtil,
+ spyRMContext, conf, 125);
+ scheduler = fs;
+
+ ConcurrentMap<ApplicationId, RMApp> spyApps =
+ spy(new ConcurrentHashMap<ApplicationId, RMApp>());
+ RMApp rmApp = mock(RMApp.class);
+ when(rmApp.getRMAppAttempt((ApplicationAttemptId) Matchers.any()))
+ .thenReturn(null);
+ Mockito.doReturn(rmApp).when(spyApps).get((ApplicationId) Matchers.any());
+ when(spyRMContext.getRMApps()).thenReturn(spyApps);
+
+ ReservationSystemTestUtil.setupFSAllocationFile(ALLOC_FILE);
+ setupPlanFollower();
+ }
+
+ private void setupPlanFollower() throws Exception {
+ ReservationSystemTestUtil testUtil = new ReservationSystemTestUtil();
+ mClock = mock(Clock.class);
+ mAgent = mock(ReservationAgent.class);
+
+ String reservationQ = testUtil.getFullReservationQueueName();
+ AllocationConfiguration allocConf = fs.getAllocationConfiguration();
+ allocConf.setReservationWindow(20L);
+ allocConf.setAverageCapacity(20);
+ policy.init(reservationQ, allocConf);
+ }
+
+ @Test
+ public void testWithMoveOnExpiry() throws PlanningException,
+ InterruptedException, AccessControlException {
+ // invoke plan follower test with move
+ testPlanFollower(true);
+ }
+
+ @Test
+ public void testWithKillOnExpiry() throws PlanningException,
+ InterruptedException, AccessControlException {
+ // invoke plan follower test with kill
+ testPlanFollower(false);
+ }
+
+ @Override
+ protected void verifyCapacity(Queue defQ) {
+ assertTrue(((FSQueue) defQ).getWeights().getWeight(ResourceType.MEMORY) >
+ 0.9);
+ }
+
+ @Override
+ protected Queue getDefaultQueue() {
+ return getReservationQueue("dedicated" +
+ ReservationConstants.DEFAULT_QUEUE_SUFFIX);
+ }
+
+ @Override
+ protected int getNumberOfApplications(Queue queue) {
+ int numberOfApplications = fs.getAppsInQueue(queue.getQueueName()).size();
+ return numberOfApplications;
+ }
+
+ @Override
+ protected AbstractSchedulerPlanFollower createPlanFollower() {
+ FairSchedulerPlanFollower planFollower =
+ new FairSchedulerPlanFollower();
+ planFollower.init(mClock, scheduler, Collections.singletonList(plan));
+ return planFollower;
+ }
+
+ @Override
+ protected void assertReservationQueueExists(ReservationId r) {
+ Queue q = getReservationQueue(r.toString());
+ assertNotNull(q);
+ }
+
+ @Override
+ protected void assertReservationQueueExists(ReservationId r,
+ double expectedCapacity, double expectedMaxCapacity) {
+ FSLeafQueue q = fs.getQueueManager().getLeafQueue(plan.getQueueName() + "" +
+ "." +
+ r, false);
+ assertNotNull(q);
+ // For now we are setting both to same weight
+ Assert.assertEquals(expectedCapacity, q.getWeights().getWeight
+ (ResourceType.MEMORY), 0.01);
+ }
+
+ @Override
+ protected void assertReservationQueueDoesNotExist(ReservationId r) {
+ Queue q = getReservationQueue(r.toString());
+ assertNull(q);
+ }
+
+ @Override
+ protected Queue getReservationQueue(String r) {
+ return fs.getQueueManager().getLeafQueue(plan.getQueueName() + "" +
+ "." +
+ r, false);
+ }
+
+ public static ApplicationACLsManager mockAppACLsManager() {
+ Configuration conf = new Configuration();
+ return new ApplicationACLsManager(conf);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (scheduler != null) {
+ fs.stop();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/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
index 73d8a55..ce3382f 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/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
@@ -29,6 +29,7 @@ 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.reservation.ReservationConstants;
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;
@@ -217,7 +218,7 @@ public class TestCapacitySchedulerDynamicBehavior {
assertEquals(1, appsInRoot.size());
// create the default reservation queue
- String defQName = "a" + PlanQueue.DEFAULT_QUEUE_SUFFIX;
+ String defQName = "a" + ReservationConstants.DEFAULT_QUEUE_SUFFIX;
ReservationQueue defQ =
new ReservationQueue(scheduler, defQName,
(PlanQueue) scheduler.getQueue("a"));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairSchedulerTestBase.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/fair/FairSchedulerTestBase.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairSchedulerTestBase.java
index 7b6aaf3..8656175 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairSchedulerTestBase.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairSchedulerTestBase.java
@@ -60,7 +60,7 @@ public class FairSchedulerTestBase {
}
}
- protected final static String TEST_DIR =
+ public final static String TEST_DIR =
new File(System.getProperty("test.build.data", "/tmp")).getAbsolutePath();
private static RecordFactory
@@ -74,7 +74,7 @@ public class FairSchedulerTestBase {
protected ResourceManager resourceManager;
// Helper methods
- protected Configuration createConfiguration() {
+ public Configuration createConfiguration() {
Configuration conf = new YarnConfiguration();
conf.setClass(YarnConfiguration.RM_SCHEDULER, FairScheduler.class,
ResourceScheduler.class);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0c4b1126/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairSchedulerPreemption.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/fair/TestFairSchedulerPreemption.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairSchedulerPreemption.java
index 903c7af..458b06d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairSchedulerPreemption.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairSchedulerPreemption.java
@@ -58,8 +58,7 @@ public class TestFairSchedulerPreemption extends FairSchedulerTestBase {
}
}
- @Override
- protected Configuration createConfiguration() {
+ public Configuration createConfiguration() {
Configuration conf = super.createConfiguration();
conf.setClass(YarnConfiguration.RM_SCHEDULER, StubbedFairScheduler.class,
ResourceScheduler.class);