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 ji...@apache.org on 2015/10/27 22:47:16 UTC
[05/50] hadoop git commit: YARN-3739. Add reservation system recovery
to RM recovery process. Contributed by Subru Krishnan.
YARN-3739. Add reservation system recovery to RM recovery process. Contributed by Subru Krishnan.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/2798723a
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/2798723a
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/2798723a
Branch: refs/heads/HDFS-8966
Commit: 2798723a5443d04455b9d79c48d61f435ab52267
Parents: 381610d
Author: Anubhav Dhoot <ad...@apache.org>
Authored: Thu Oct 22 06:36:58 2015 -0700
Committer: Anubhav Dhoot <ad...@apache.org>
Committed: Thu Oct 22 06:51:00 2015 -0700
----------------------------------------------------------------------
hadoop-yarn-project/CHANGES.txt | 3 +
.../server/resourcemanager/ClientRMService.java | 2 +-
.../server/resourcemanager/ResourceManager.java | 4 +
.../reservation/AbstractReservationSystem.java | 91 ++++-
.../AbstractSchedulerPlanFollower.java | 76 ++--
.../CapacitySchedulerPlanFollower.java | 16 -
.../reservation/FairSchedulerPlanFollower.java | 15 -
.../reservation/InMemoryPlan.java | 35 +-
.../resourcemanager/reservation/PlanEdit.java | 6 +-
.../reservation/PlanFollower.java | 4 +-
.../reservation/ReservationSystem.java | 7 +-
.../reservation/planning/PlanningAlgorithm.java | 2 +-
.../TestReservationSystemWithRMHA.java | 360 +++++++++++++++++--
.../reservation/ReservationSystemTestUtil.java | 1 -
.../reservation/TestCapacityOverTimePolicy.java | 29 +-
.../reservation/TestInMemoryPlan.java | 16 +-
.../reservation/TestNoOverCommitPolicy.java | 19 +-
.../TestSchedulerPlanFollowerBase.java | 17 +-
.../planning/TestAlignedPlanner.java | 7 +-
.../planning/TestGreedyReservationAgent.java | 17 +-
.../planning/TestSimpleCapacityReplanner.java | 14 +-
21 files changed, 556 insertions(+), 185 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index ae26386..79df1ce 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -238,6 +238,9 @@ Release 2.8.0 - UNRELEASED
YARN-4262. Allow whitelisted users to run privileged docker containers.
(Sidharta Seethana via vvasudev)
+ YARN-3739. Add reservation system recovery to RM recovery process.
+ (Subru Krishnan via adhoot)
+
IMPROVEMENTS
YARN-644. Basic null check is not performed on passed in arguments before
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.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/ClientRMService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java
index 4a02580..812267d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java
@@ -1363,7 +1363,7 @@ public class ClientRMService extends AbstractService implements
.format(
"Reservation {0} is within threshold so attempting to create synchronously.",
reservationId));
- reservationSystem.synchronizePlan(planName);
+ reservationSystem.synchronizePlan(planName, true);
LOG.info(MessageFormat.format("Created reservation {0} synchronously.",
reservationId));
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.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/ResourceManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java
index b38f188..88fb1cf 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java
@@ -1186,6 +1186,10 @@ public class ResourceManager extends CompositeService implements Recoverable {
// recover AMRMTokenSecretManager
rmContext.getAMRMTokenSecretManager().recover(state);
+ // recover reservations
+ if (reservationSystem != null) {
+ reservationSystem.recover(state);
+ }
// recover applications
rmAppManager.recover(state);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/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 cf57dbe..56423e2 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
@@ -18,16 +18,6 @@
package org.apache.hadoop.yarn.server.resourcemanager.reservation;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
import org.apache.hadoop.classification.InterfaceAudience.LimitedPrivate;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.conf.Configuration;
@@ -38,8 +28,11 @@ 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.exceptions.YarnRuntimeException;
+import org.apache.hadoop.yarn.proto.YarnServerResourceManagerRecoveryProtos.ReservationAllocationStateProto;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
+import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.PlanningException;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.planning.Planner;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.planning.ReservationAgent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
@@ -52,6 +45,17 @@ import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
/**
* This is the implementation of {@link ReservationSystem} based on the
* {@link ResourceScheduler}
@@ -94,6 +98,8 @@ public abstract class AbstractReservationSystem extends AbstractService
private PlanFollower planFollower;
+ private boolean isRecoveryEnabled = false;
+
/**
* Construct the service.
*
@@ -149,6 +155,49 @@ public abstract class AbstractReservationSystem extends AbstractService
Plan plan = initializePlan(planQueueName);
plans.put(planQueueName, plan);
}
+ isRecoveryEnabled = conf.getBoolean(
+ YarnConfiguration.RECOVERY_ENABLED,
+ YarnConfiguration.DEFAULT_RM_RECOVERY_ENABLED);
+ }
+
+ private void loadPlan(String planName,
+ Map<ReservationId, ReservationAllocationStateProto> reservations)
+ throws PlanningException {
+ Plan plan = plans.get(planName);
+ Resource minAllocation = getMinAllocation();
+ ResourceCalculator rescCalculator = getResourceCalculator();
+ for (Entry<ReservationId, ReservationAllocationStateProto> currentReservation : reservations
+ .entrySet()) {
+ plan.addReservation(ReservationSystemUtil.toInMemoryAllocation(planName,
+ currentReservation.getKey(), currentReservation.getValue(),
+ minAllocation, rescCalculator), true);
+ resQMap.put(currentReservation.getKey(), planName);
+ }
+ LOG.info("Recovered reservations for Plan: {}", planName);
+ }
+
+ @Override
+ public void recover(RMState state) throws Exception {
+ LOG.info("Recovering Reservation system");
+ writeLock.lock();
+ try {
+ Map<String, Map<ReservationId, ReservationAllocationStateProto>> reservationSystemState =
+ state.getReservationState();
+ if (planFollower != null) {
+ for (String plan : plans.keySet()) {
+ // recover reservations if any from state store
+ if (reservationSystemState.containsKey(plan)) {
+ loadPlan(plan, reservationSystemState.get(plan));
+ }
+ synchronizePlan(plan, false);
+ }
+ startPlanFollower(conf.getLong(
+ YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_SCHEDULING_WAIT_MS,
+ YarnConfiguration.DEFAULT_RM_WORK_PRESERVING_RECOVERY_SCHEDULING_WAIT_MS));
+ }
+ } finally {
+ writeLock.unlock();
+ }
}
private void initializeNewPlans(Configuration conf) {
@@ -162,7 +211,7 @@ public abstract class AbstractReservationSystem extends AbstractService
Plan plan = initializePlan(planQueueName);
plans.put(planQueueName, plan);
} else {
- LOG.warn("Plan based on reservation queue {0} already exists.",
+ LOG.warn("Plan based on reservation queue {} already exists.",
planQueueName);
}
}
@@ -236,18 +285,26 @@ public abstract class AbstractReservationSystem extends AbstractService
}
@Override
- public void synchronizePlan(String planName) {
+ public void synchronizePlan(String planName, boolean shouldReplan) {
writeLock.lock();
try {
Plan plan = plans.get(planName);
if (plan != null) {
- planFollower.synchronizePlan(plan);
+ planFollower.synchronizePlan(plan, shouldReplan);
}
} finally {
writeLock.unlock();
}
}
+ private void startPlanFollower(long initialDelay) {
+ if (planFollower != null) {
+ scheduledExecutorService = new ScheduledThreadPoolExecutor(1);
+ scheduledExecutorService.scheduleWithFixedDelay(planFollower,
+ initialDelay, planStepSize, TimeUnit.MILLISECONDS);
+ }
+ }
+
@Override
public void serviceInit(Configuration conf) throws Exception {
Configuration configuration = new Configuration(conf);
@@ -262,10 +319,8 @@ public abstract class AbstractReservationSystem extends AbstractService
@Override
public void serviceStart() throws Exception {
- if (planFollower != null) {
- scheduledExecutorService = new ScheduledThreadPoolExecutor(1);
- scheduledExecutorService.scheduleWithFixedDelay(planFollower, 0L,
- planStepSize, TimeUnit.MILLISECONDS);
+ if (!isRecoveryEnabled) {
+ startPlanFollower(planStepSize);
}
super.serviceStart();
}
@@ -350,7 +405,7 @@ public abstract class AbstractReservationSystem extends AbstractService
minAllocation, maxAllocation, planQueueName,
getReplanner(planQueuePath), getReservationSchedulerConfiguration()
.getMoveOnExpiry(planQueuePath), rmContext);
- LOG.info("Intialized plan {0} based on reservable queue {1}",
+ LOG.info("Intialized plan {} based on reservable queue {}",
plan.toString(), planQueueName);
return plan;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/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 ea7f27d..eaf2902 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
@@ -27,8 +27,8 @@ 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.common.QueueEntitlement;
-
import org.apache.hadoop.yarn.util.Clock;
+import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import org.apache.hadoop.yarn.util.resource.Resources;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,7 +43,7 @@ import java.util.Set;
public abstract class AbstractSchedulerPlanFollower implements PlanFollower {
private static final Logger LOG = LoggerFactory
- .getLogger(CapacitySchedulerPlanFollower.class);
+ .getLogger(AbstractSchedulerPlanFollower.class);
protected Collection<Plan> plans = new ArrayList<Plan>();
protected YarnScheduler scheduler;
@@ -59,7 +59,7 @@ public abstract class AbstractSchedulerPlanFollower implements PlanFollower {
@Override
public synchronized void run() {
for (Plan plan : plans) {
- synchronizePlan(plan);
+ synchronizePlan(plan, true);
}
}
@@ -70,7 +70,7 @@ public abstract class AbstractSchedulerPlanFollower implements PlanFollower {
}
@Override
- public synchronized void synchronizePlan(Plan plan) {
+ public synchronized void synchronizePlan(Plan plan, boolean shouldReplan) {
String planQueueName = plan.getQueueName();
if (LOG.isDebugEnabled()) {
LOG.debug("Running plan follower edit policy for plan: " + planQueueName);
@@ -88,14 +88,12 @@ public abstract class AbstractSchedulerPlanFollower implements PlanFollower {
Resource clusterResources = scheduler.getClusterResource();
Resource planResources = getPlanResources(plan, planQueue,
clusterResources);
-
Set<ReservationAllocation> currentReservations =
plan.getReservationsAtTime(now);
Set<String> curReservationNames = new HashSet<String>();
Resource reservedResources = Resource.newInstance(0, 0);
int numRes = getReservedResources(now, currentReservations,
curReservationNames, reservedResources);
-
// create the default reservation queue if it doesnt exist
String defReservationId = getReservationIdFromQueueName(planQueueName) +
ReservationConstants.DEFAULT_QUEUE_SUFFIX;
@@ -104,14 +102,18 @@ public abstract class AbstractSchedulerPlanFollower implements PlanFollower {
createDefaultReservationQueue(planQueueName, planQueue,
defReservationId);
curReservationNames.add(defReservationId);
-
// if the resources dedicated to this plan has shrunk invoke replanner
- if (arePlanResourcesLessThanReservations(clusterResources, planResources,
- reservedResources)) {
- try {
- plan.getReplanner().plan(plan, null);
- } catch (PlanningException e) {
- LOG.warn("Exception while trying to replan: {}", planQueueName, e);
+ boolean shouldResize = false;
+ if (arePlanResourcesLessThanReservations(plan.getResourceCalculator(),
+ clusterResources, planResources, reservedResources)) {
+ if (shouldReplan) {
+ try {
+ plan.getReplanner().plan(plan, null);
+ } catch (PlanningException e) {
+ LOG.warn("Exception while trying to replan: {}", planQueueName, e);
+ }
+ } else {
+ shouldResize = true;
}
}
// identify the reservations that have expired and new reservations that
@@ -133,7 +135,6 @@ public abstract class AbstractSchedulerPlanFollower implements PlanFollower {
// garbage collect expired reservations
cleanupExpiredQueues(planQueueName, plan.getMoveOnExpiry(), expired,
defReservationQueue);
-
// Add new reservations and update existing ones
float totalAssignedCapacity = 0f;
if (currentReservations != null) {
@@ -146,9 +147,8 @@ public abstract class AbstractSchedulerPlanFollower implements PlanFollower {
planQueueName, e);
}
// sort allocations from the one giving up the most resources, to the
- // one asking for the most
- // avoid order-of-operation errors that temporarily violate 100%
- // capacity bound
+ // one asking for the most avoid order-of-operation errors that
+ // temporarily violate 100% capacity bound
List<ReservationAllocation> sortedAllocations =
sortByDelta(
new ArrayList<ReservationAllocation>(currentReservations), now,
@@ -162,10 +162,15 @@ public abstract class AbstractSchedulerPlanFollower implements PlanFollower {
float targetCapacity = 0f;
if (planResources.getMemory() > 0
&& planResources.getVirtualCores() > 0) {
+ if (shouldResize) {
+ capToAssign =
+ calculateReservationToPlanProportion(
+ plan.getResourceCalculator(), planResources,
+ reservedResources, capToAssign);
+ }
targetCapacity =
- calculateReservationToPlanRatio(clusterResources,
- planResources,
- capToAssign);
+ calculateReservationToPlanRatio(plan.getResourceCalculator(),
+ clusterResources, planResources, capToAssign);
}
if (LOG.isDebugEnabled()) {
LOG.debug(
@@ -211,7 +216,6 @@ public abstract class AbstractSchedulerPlanFollower implements PlanFollower {
}
LOG.info("Finished iteration of plan follower edit policy for plan: "
+ planQueueName);
-
// Extension: update plan with app states,
// useful to support smart replanning
}
@@ -324,18 +328,34 @@ public abstract class AbstractSchedulerPlanFollower implements PlanFollower {
protected abstract Queue getPlanQueue(String planQueueName);
/**
+ * Resizes reservations based on currently available resources
+ */
+ private Resource calculateReservationToPlanProportion(
+ ResourceCalculator rescCalculator, Resource availablePlanResources,
+ Resource totalReservationResources, Resource reservationResources) {
+ return Resources.multiply(availablePlanResources, Resources.ratio(
+ rescCalculator, reservationResources, totalReservationResources));
+ }
+
+ /**
* Calculates ratio of reservationResources to planResources
*/
- protected abstract float calculateReservationToPlanRatio(
- Resource clusterResources, Resource planResources,
- Resource reservationResources);
+ private float calculateReservationToPlanRatio(
+ ResourceCalculator rescCalculator, Resource clusterResources,
+ Resource planResources, Resource reservationResources) {
+ return Resources.divide(rescCalculator, clusterResources,
+ reservationResources, planResources);
+ }
/**
* Check if plan resources are less than expected reservation resources
*/
- protected abstract boolean arePlanResourcesLessThanReservations(
- Resource clusterResources, Resource planResources,
- Resource reservedResources);
+ private boolean arePlanResourcesLessThanReservations(
+ ResourceCalculator rescCalculator, Resource clusterResources,
+ Resource planResources, Resource reservedResources) {
+ return Resources.greaterThan(rescCalculator, clusterResources,
+ reservedResources, planResources);
+ }
/**
* Get a list of reservation queues for this planQueue
@@ -363,7 +383,7 @@ public abstract class AbstractSchedulerPlanFollower implements PlanFollower {
Plan plan, Queue queue, Resource clusterResources);
/**
- * Get reservation queue resources if it exists otherwise return null
+ * Get reservation queue resources if it exists otherwise return null.
*/
protected abstract Resource getReservationQueueResourceIfExists(Plan plan,
ReservationId reservationId);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/CapacitySchedulerPlanFollower.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/CapacitySchedulerPlanFollower.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/CapacitySchedulerPlanFollower.java
index 61772c9..551f075 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/CapacitySchedulerPlanFollower.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/CapacitySchedulerPlanFollower.java
@@ -81,22 +81,6 @@ public class CapacitySchedulerPlanFollower extends AbstractSchedulerPlanFollower
}
@Override
- protected float calculateReservationToPlanRatio(
- Resource clusterResources, Resource planResources,
- Resource reservationResources) {
- return Resources.divide(cs.getResourceCalculator(),
- clusterResources, reservationResources, planResources);
- }
-
- @Override
- protected boolean arePlanResourcesLessThanReservations(
- Resource clusterResources, Resource planResources,
- Resource reservedResources) {
- return Resources.greaterThan(cs.getResourceCalculator(),
- clusterResources, reservedResources, planResources);
- }
-
- @Override
protected List<? extends Queue> getChildReservationQueues(Queue queue) {
PlanQueue planQueue = (PlanQueue)queue;
List<CSQueue> childQueues = planQueue.getChildQueues();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/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
index 7ca03c5..2a57b22 100644
--- 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
@@ -30,7 +30,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSParentQueu
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;
@@ -60,20 +59,6 @@ public class FairSchedulerPlanFollower extends AbstractSchedulerPlanFollower {
}
@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();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/InMemoryPlan.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/InMemoryPlan.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/InMemoryPlan.java
index a887e24..7e2567b 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/InMemoryPlan.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/InMemoryPlan.java
@@ -33,6 +33,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.hadoop.yarn.api.records.ReservationId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.PlanningException;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.planning.Planner;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.planning.ReservationAgent;
@@ -54,7 +55,7 @@ public class InMemoryPlan implements Plan {
private static final Logger LOG = LoggerFactory.getLogger(InMemoryPlan.class);
private static final Resource ZERO_RESOURCE = Resource.newInstance(0, 0);
- private final RMContext rmContext;
+ private final RMStateStore rmStateStore;
private TreeMap<ReservationInterval, Set<InMemoryReservationAllocation>> currentReservations =
new TreeMap<ReservationInterval, Set<InMemoryReservationAllocation>>();
@@ -112,7 +113,7 @@ public class InMemoryPlan implements Plan {
this.replanner = replanner;
this.getMoveOnExpiry = getMoveOnExpiry;
this.clock = clock;
- this.rmContext = rmContext;
+ this.rmStateStore = rmContext.getStateStore();
}
@Override
@@ -174,8 +175,8 @@ public class InMemoryPlan implements Plan {
}
@Override
- public boolean addReservation(ReservationAllocation reservation)
- throws PlanningException {
+ public boolean addReservation(ReservationAllocation reservation,
+ boolean isRecovering) throws PlanningException {
// Verify the allocation is memory based otherwise it is not supported
InMemoryReservationAllocation inMemReservation =
(InMemoryReservationAllocation) reservation;
@@ -198,9 +199,16 @@ public class InMemoryPlan implements Plan {
}
// Validate if we can accept this reservation, throws exception if
// validation fails
- policy.validate(this, inMemReservation);
- // we record here the time in which the allocation has been accepted
- reservation.setAcceptanceTimestamp(clock.getTime());
+ if (!isRecovering) {
+ policy.validate(this, inMemReservation);
+ // we record here the time in which the allocation has been accepted
+ reservation.setAcceptanceTimestamp(clock.getTime());
+ if (rmStateStore != null) {
+ rmStateStore.storeNewReservation(
+ ReservationSystemUtil.buildStateProto(inMemReservation),
+ getQueueName(), inMemReservation.getReservationId().toString());
+ }
+ }
ReservationInterval searchInterval =
new ReservationInterval(inMemReservation.getStartTime(),
inMemReservation.getEndTime());
@@ -217,9 +225,6 @@ public class InMemoryPlan implements Plan {
currentReservations.put(searchInterval, reservations);
reservationTable.put(inMemReservation.getReservationId(),
inMemReservation);
- rmContext.getStateStore().storeNewReservation(
- ReservationSystemUtil.buildStateProto(inMemReservation),
- getQueueName(), inMemReservation.getReservationId().toString());
incrementAllocation(inMemReservation);
LOG.info("Sucessfully added reservation: {} to plan.",
inMemReservation.getReservationId());
@@ -253,7 +258,7 @@ public class InMemoryPlan implements Plan {
return result;
}
try {
- result = addReservation(reservation);
+ result = addReservation(reservation, false);
} catch (PlanningException e) {
LOG.error("Unable to update reservation: {} from plan due to {}.",
reservation.getReservationId(), e.getMessage());
@@ -264,7 +269,7 @@ public class InMemoryPlan implements Plan {
return result;
} else {
// rollback delete
- addReservation(currReservation);
+ addReservation(currReservation, false);
LOG.info("Rollbacked update reservation: {} from plan.",
reservation.getReservationId());
return result;
@@ -282,6 +287,10 @@ public class InMemoryPlan implements Plan {
Set<InMemoryReservationAllocation> reservations =
currentReservations.get(searchInterval);
if (reservations != null) {
+ if (rmStateStore != null) {
+ rmStateStore.removeReservation(getQueueName(),
+ reservation.getReservationId().toString());
+ }
if (!reservations.remove(reservation)) {
LOG.error("Unable to remove reservation: {} from plan.",
reservation.getReservationId());
@@ -298,8 +307,6 @@ public class InMemoryPlan implements Plan {
throw new IllegalArgumentException(errMsg);
}
reservationTable.remove(reservation.getReservationId());
- rmContext.getStateStore().removeReservation(
- getQueueName(), reservation.getReservationId().toString());
decrementAllocation(reservation);
LOG.info("Sucessfully deleted reservation: {} in plan.",
reservation.getReservationId());
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/PlanEdit.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/PlanEdit.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/PlanEdit.java
index 60e201b..504a250 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/PlanEdit.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/PlanEdit.java
@@ -32,10 +32,12 @@ public interface PlanEdit extends PlanContext, PlanView {
*
* @param reservation the {@link ReservationAllocation} to be added to the
* plan
+ * @param isRecovering flag to indicate if reservation is being added as part
+ * of failover or not
* @return true if addition is successful, false otherwise
*/
- public boolean addReservation(ReservationAllocation reservation)
- throws PlanningException;
+ public boolean addReservation(ReservationAllocation reservation,
+ boolean isRecovering) throws PlanningException;
/**
* Updates an existing {@link ReservationAllocation} in the plan. This is
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/PlanFollower.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/PlanFollower.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/PlanFollower.java
index 6635314..a5c8ee1 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/PlanFollower.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/PlanFollower.java
@@ -71,8 +71,10 @@ public interface PlanFollower extends Runnable {
* start time is imminent.
*
* @param plan the Plan to synchronize
+ * @param shouldReplan replan on reduction of plan capacity if true or
+ * proportionally scale down reservations if false
*/
- public void synchronizePlan(Plan plan);
+ public void synchronizePlan(Plan plan, boolean shouldReplan);
/**
* Setter for the list of plans.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationSystem.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/ReservationSystem.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationSystem.java
index 7785885..56a08ef 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationSystem.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/ReservationSystem.java
@@ -24,6 +24,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.api.records.ReservationId;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.Recoverable;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Queue;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
@@ -40,7 +41,7 @@ import java.util.Map;
*/
@LimitedPrivate("yarn")
@Unstable
-public interface ReservationSystem {
+public interface ReservationSystem extends Recoverable {
/**
* Set RMContext for {@link ReservationSystem}. This method should be called
@@ -82,8 +83,10 @@ public interface ReservationSystem {
* the {@link ResourceScheduler}
*
* @param planName the name of the {@link Plan} to be synchronized
+ * @param shouldReplan replan on reduction of plan capacity if true or
+ * proportionally scale down reservations if false
*/
- void synchronizePlan(String planName);
+ void synchronizePlan(String planName, boolean shouldReplan);
/**
* Return the time step (ms) at which the {@link PlanFollower} is invoked
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/PlanningAlgorithm.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/planning/PlanningAlgorithm.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/PlanningAlgorithm.java
index 9a0a0f0..8b72b9f 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/PlanningAlgorithm.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/PlanningAlgorithm.java
@@ -94,7 +94,7 @@ public abstract class PlanningAlgorithm implements ReservationAgent {
if (oldReservation != null) {
return plan.updateReservation(capReservation);
} else {
- return plan.addReservation(capReservation);
+ return plan.addReservation(capReservation, false);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestReservationSystemWithRMHA.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/TestReservationSystemWithRMHA.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestReservationSystemWithRMHA.java
index 6f74130..48a4d97 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestReservationSystemWithRMHA.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestReservationSystemWithRMHA.java
@@ -18,15 +18,20 @@
package org.apache.hadoop.yarn.server.resourcemanager;
import org.apache.hadoop.yarn.api.protocolrecords.ReservationDeleteRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.ReservationDeleteResponse;
import org.apache.hadoop.yarn.api.protocolrecords.ReservationSubmissionRequest;
import org.apache.hadoop.yarn.api.protocolrecords.ReservationSubmissionResponse;
import org.apache.hadoop.yarn.api.protocolrecords.ReservationUpdateRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.ReservationUpdateResponse;
+import org.apache.hadoop.yarn.api.records.QueueInfo;
import org.apache.hadoop.yarn.api.records.ReservationDefinition;
import org.apache.hadoop.yarn.api.records.ReservationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.event.DrainDispatcher;
import org.apache.hadoop.yarn.proto.YarnServerResourceManagerRecoveryProtos.ReservationAllocationStateProto;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
+import org.apache.hadoop.yarn.server.resourcemanager.reservation.Plan;
+import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationAllocation;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSystemTestUtil;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
@@ -38,7 +43,7 @@ import org.junit.Test;
import java.util.Map;
-public class TestReservationSystemWithRMHA extends RMHATestBase{
+public class TestReservationSystemWithRMHA extends RMHATestBase {
@Override
public void setup() throws Exception {
@@ -56,7 +61,7 @@ public class TestReservationSystemWithRMHA extends RMHATestBase{
public void testSubmitReservationAndCheckAfterFailover() throws Exception {
startRMs();
- addNodeCapacityToPlan();
+ addNodeCapacityToPlan(rm1, 102400, 100);
ClientRMService clientService = rm1.getClientRMService();
@@ -72,8 +77,6 @@ public class TestReservationSystemWithRMHA extends RMHATestBase{
ReservationId reservationID = response.getReservationId();
Assert.assertNotNull(reservationID);
LOG.info("Submit reservation response: " + reservationID);
- ReservationDefinition reservationDefinition = request
- .getReservationDefinition();
// Do the failover
explicitFailover();
@@ -87,12 +90,11 @@ public class TestReservationSystemWithRMHA extends RMHATestBase{
Assert.assertNotNull(reservationStateMap.get(reservationID));
}
-
@Test
public void testUpdateReservationAndCheckAfterFailover() throws Exception {
startRMs();
- addNodeCapacityToPlan();
+ addNodeCapacityToPlan(rm1, 102400, 100);
ClientRMService clientService = rm1.getClientRMService();
@@ -108,17 +110,15 @@ public class TestReservationSystemWithRMHA extends RMHATestBase{
ReservationId reservationID = response.getReservationId();
Assert.assertNotNull(reservationID);
LOG.info("Submit reservation response: " + reservationID);
- ReservationDefinition reservationDefinition = request
- .getReservationDefinition();
-
+ ReservationDefinition reservationDefinition =
+ request.getReservationDefinition();
// Change any field
long newDeadline = reservationDefinition.getDeadline() + 100;
reservationDefinition.setDeadline(newDeadline);
- ReservationUpdateRequest updateRequest =
- ReservationUpdateRequest.newInstance(
- reservationDefinition, reservationID);
+ ReservationUpdateRequest updateRequest = ReservationUpdateRequest
+ .newInstance(reservationDefinition, reservationID);
rm1.updateReservationState(updateRequest);
// Do the failover
@@ -140,7 +140,7 @@ public class TestReservationSystemWithRMHA extends RMHATestBase{
public void testDeleteReservationAndCheckAfterFailover() throws Exception {
startRMs();
- addNodeCapacityToPlan();
+ addNodeCapacityToPlan(rm1, 102400, 100);
ClientRMService clientService = rm1.getClientRMService();
@@ -156,7 +156,6 @@ public class TestReservationSystemWithRMHA extends RMHATestBase{
ReservationId reservationID = response.getReservationId();
Assert.assertNotNull(reservationID);
-
// Delete the reservation
ReservationDeleteRequest deleteRequest =
ReservationDeleteRequest.newInstance(reservationID);
@@ -168,32 +167,31 @@ public class TestReservationSystemWithRMHA extends RMHATestBase{
rm2.registerNode("127.0.0.1:1", 102400, 100);
RMState state = rm2.getRMContext().getStateStore().loadState();
- Assert.assertNull(state.getReservationState().get(
- ReservationSystemTestUtil.reservationQ));
+ Assert.assertNull(state.getReservationState()
+ .get(ReservationSystemTestUtil.reservationQ));
}
- private void addNodeCapacityToPlan() {
+ private void addNodeCapacityToPlan(MockRM rm, int memory, int vCores) {
try {
- rm1.registerNode("127.0.0.1:1", 102400, 100);
+ rm.registerNode("127.0.0.1:1", memory, vCores);
int attempts = 10;
do {
DrainDispatcher dispatcher =
(DrainDispatcher) rm1.getRMContext().getDispatcher();
dispatcher.await();
- rm1.getRMContext().getReservationSystem().synchronizePlan(
- ReservationSystemTestUtil.reservationQ);
- if (rm1.getRMContext().getReservationSystem().getPlan
- (ReservationSystemTestUtil.reservationQ).getTotalCapacity()
+ rm.getRMContext().getReservationSystem()
+ .synchronizePlan(ReservationSystemTestUtil.reservationQ, false);
+ if (rm.getRMContext().getReservationSystem()
+ .getPlan(ReservationSystemTestUtil.reservationQ).getTotalCapacity()
.getMemory() > 0) {
break;
}
LOG.info("Waiting for node capacity to be added to plan");
Thread.sleep(100);
- }
- while (attempts-- > 0);
+ } while (attempts-- > 0);
if (attempts <= 0) {
- Assert.fail("Exhausted attempts in checking if node capacity was " +
- "added to the plan");
+ Assert.fail("Exhausted attempts in checking if node capacity was "
+ + "added to the plan");
}
} catch (Exception e) {
@@ -205,8 +203,316 @@ public class TestReservationSystemWithRMHA extends RMHATestBase{
Clock clock = new UTCClock();
long arrival = clock.getTime();
long duration = 60000;
- long deadline = (long) (arrival + 1.05 * duration);
+ long deadline = (long) (arrival + duration + 1500);
return ReservationSystemTestUtil.createSimpleReservationRequest(4, arrival,
deadline, duration);
}
+
+ private void validateReservation(Plan plan, ReservationId resId,
+ ReservationDefinition rDef) {
+ ReservationAllocation reservation = plan.getReservationById(resId);
+ Assert.assertNotNull(reservation);
+ Assert.assertEquals(rDef.getDeadline(),
+ reservation.getReservationDefinition().getDeadline());
+ }
+
+ @Test
+ public void testSubmitReservationFailoverAndDelete() throws Exception {
+ startRMs();
+
+ addNodeCapacityToPlan(rm1, 102400, 100);
+
+ ClientRMService clientService = rm1.getClientRMService();
+
+ // create a reservation
+ ReservationSubmissionRequest request = createReservationSubmissionRequest();
+ ReservationSubmissionResponse response = null;
+ try {
+ response = clientService.submitReservation(request);
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ Assert.assertNotNull(response);
+ ReservationId reservationID = response.getReservationId();
+ Assert.assertNotNull(reservationID);
+ LOG.info("Submit reservation response: " + reservationID);
+ ReservationDefinition reservationDefinition =
+ request.getReservationDefinition();
+
+ // Do the failover
+ explicitFailover();
+
+ addNodeCapacityToPlan(rm2, 102400, 100);
+
+ // check if reservation exists after failover
+ Plan plan = rm2.getRMContext().getReservationSystem()
+ .getPlan(ReservationSystemTestUtil.reservationQ);
+ validateReservation(plan, reservationID, reservationDefinition);
+
+ // delete the reservation
+ ReservationDeleteRequest deleteRequest =
+ ReservationDeleteRequest.newInstance(reservationID);
+ ReservationDeleteResponse deleteResponse = null;
+ clientService = rm2.getClientRMService();
+ try {
+ deleteResponse = clientService.deleteReservation(deleteRequest);
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ Assert.assertNotNull(deleteResponse);
+ Assert.assertNull(plan.getReservationById(reservationID));
+ }
+
+ @Test
+ public void testFailoverAndSubmitReservation() throws Exception {
+ startRMs();
+
+ addNodeCapacityToPlan(rm1, 102400, 100);
+
+ // Do the failover
+ explicitFailover();
+
+ addNodeCapacityToPlan(rm2, 102400, 100);
+
+ // create a reservation
+ ClientRMService clientService = rm2.getClientRMService();
+ ReservationSubmissionRequest request = createReservationSubmissionRequest();
+ ReservationSubmissionResponse response = null;
+ try {
+ response = clientService.submitReservation(request);
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ Assert.assertNotNull(response);
+ ReservationId reservationID = response.getReservationId();
+ Assert.assertNotNull(reservationID);
+ LOG.info("Submit reservation response: " + reservationID);
+ ReservationDefinition reservationDefinition =
+ request.getReservationDefinition();
+
+ // check if reservation is submitted successfully
+ Plan plan = rm2.getRMContext().getReservationSystem()
+ .getPlan(ReservationSystemTestUtil.reservationQ);
+ validateReservation(plan, reservationID, reservationDefinition);
+ }
+
+ @Test
+ public void testSubmitReservationFailoverAndUpdate() throws Exception {
+ startRMs();
+
+ addNodeCapacityToPlan(rm1, 102400, 100);
+
+ ClientRMService clientService = rm1.getClientRMService();
+
+ // create a reservation
+ ReservationSubmissionRequest request = createReservationSubmissionRequest();
+ ReservationSubmissionResponse response = null;
+ try {
+ response = clientService.submitReservation(request);
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ Assert.assertNotNull(response);
+ ReservationId reservationID = response.getReservationId();
+ Assert.assertNotNull(reservationID);
+ LOG.info("Submit reservation response: " + reservationID);
+ ReservationDefinition reservationDefinition =
+ request.getReservationDefinition();
+
+ // Do the failover
+ explicitFailover();
+
+ addNodeCapacityToPlan(rm2, 102400, 100);
+
+ // check if reservation exists after failover
+ Plan plan = rm2.getRMContext().getReservationSystem()
+ .getPlan(ReservationSystemTestUtil.reservationQ);
+ validateReservation(plan, reservationID, reservationDefinition);
+
+ // update the reservation
+ long newDeadline = reservationDefinition.getDeadline() + 100;
+ reservationDefinition.setDeadline(newDeadline);
+ ReservationUpdateRequest updateRequest = ReservationUpdateRequest
+ .newInstance(reservationDefinition, reservationID);
+ ReservationUpdateResponse updateResponse = null;
+ clientService = rm2.getClientRMService();
+ try {
+ updateResponse = clientService.updateReservation(updateRequest);
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ Assert.assertNotNull(updateResponse);
+ validateReservation(plan, reservationID, reservationDefinition);
+ }
+
+ @Test
+ public void testSubmitUpdateReservationFailoverAndDelete() throws Exception {
+ startRMs();
+
+ addNodeCapacityToPlan(rm1, 102400, 100);
+
+ ClientRMService clientService = rm1.getClientRMService();
+
+ // create a reservation
+ ReservationSubmissionRequest request = createReservationSubmissionRequest();
+ ReservationSubmissionResponse response = null;
+ try {
+ response = clientService.submitReservation(request);
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ Assert.assertNotNull(response);
+ ReservationId reservationID = response.getReservationId();
+ Assert.assertNotNull(reservationID);
+ LOG.info("Submit reservation response: " + reservationID);
+ ReservationDefinition reservationDefinition =
+ request.getReservationDefinition();
+
+ // check if reservation is submitted successfully
+ Plan plan = rm1.getRMContext().getReservationSystem()
+ .getPlan(ReservationSystemTestUtil.reservationQ);
+ validateReservation(plan, reservationID, reservationDefinition);
+
+ // update the reservation
+ long newDeadline = reservationDefinition.getDeadline() + 100;
+ reservationDefinition.setDeadline(newDeadline);
+ ReservationUpdateRequest updateRequest = ReservationUpdateRequest
+ .newInstance(reservationDefinition, reservationID);
+ ReservationUpdateResponse updateResponse = null;
+ try {
+ updateResponse = clientService.updateReservation(updateRequest);
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ Assert.assertNotNull(updateResponse);
+ validateReservation(plan, reservationID, reservationDefinition);
+
+ // Do the failover
+ explicitFailover();
+
+ addNodeCapacityToPlan(rm2, 102400, 100);
+
+ // check if reservation exists after failover
+ plan = rm2.getRMContext().getReservationSystem()
+ .getPlan(ReservationSystemTestUtil.reservationQ);
+ validateReservation(plan, reservationID, reservationDefinition);
+
+ // delete the reservation
+ ReservationDeleteRequest deleteRequest =
+ ReservationDeleteRequest.newInstance(reservationID);
+ ReservationDeleteResponse deleteResponse = null;
+ clientService = rm2.getClientRMService();
+ try {
+ deleteResponse = clientService.deleteReservation(deleteRequest);
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ Assert.assertNotNull(deleteResponse);
+ Assert.assertNull(plan.getReservationById(reservationID));
+ }
+
+ @Test
+ public void testReservationResizeAfterFailover() throws Exception {
+ startRMs();
+
+ addNodeCapacityToPlan(rm1, 102400, 100);
+
+ ClientRMService clientService = rm1.getClientRMService();
+
+ // create 3 reservations
+ ReservationSubmissionRequest request = createReservationSubmissionRequest();
+ ReservationDefinition reservationDefinition =
+ request.getReservationDefinition();
+ ReservationSubmissionResponse response = null;
+ try {
+ response = clientService.submitReservation(request);
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ Assert.assertNotNull(response);
+ ReservationId resID1 = response.getReservationId();
+ Assert.assertNotNull(resID1);
+ LOG.info("Submit reservation response: " + resID1);
+ try {
+ response = clientService.submitReservation(request);
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ Assert.assertNotNull(response);
+ ReservationId resID2 = response.getReservationId();
+ Assert.assertNotNull(resID2);
+ LOG.info("Submit reservation response: " + resID2);
+ try {
+ response = clientService.submitReservation(request);
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ Assert.assertNotNull(response);
+ ReservationId resID3 = response.getReservationId();
+ Assert.assertNotNull(resID3);
+ LOG.info("Submit reservation response: " + resID3);
+
+ // allow the reservations to become active
+ waitForReservationActivation(rm1, resID1,
+ ReservationSystemTestUtil.reservationQ);
+
+ // validate reservations before failover
+ Plan plan = rm1.getRMContext().getReservationSystem()
+ .getPlan(ReservationSystemTestUtil.reservationQ);
+ validateReservation(plan, resID1, reservationDefinition);
+ validateReservation(plan, resID2, reservationDefinition);
+ validateReservation(plan, resID3, reservationDefinition);
+ ResourceScheduler scheduler = rm1.getResourceScheduler();
+ QueueInfo resQ1 = scheduler.getQueueInfo(resID1.toString(), false, false);
+ Assert.assertEquals(0.05, resQ1.getCapacity(), 0.001f);
+ QueueInfo resQ2 = scheduler.getQueueInfo(resID2.toString(), false, false);
+ Assert.assertEquals(0.05, resQ2.getCapacity(), 0.001f);
+ QueueInfo resQ3 = scheduler.getQueueInfo(resID3.toString(), false, false);
+ Assert.assertEquals(0.05, resQ3.getCapacity(), 0.001f);
+
+ // Do the failover
+ explicitFailover();
+
+ addNodeCapacityToPlan(rm2, 5120, 5);
+
+ // check if reservations exists after failover
+ plan = rm2.getRMContext().getReservationSystem()
+ .getPlan(ReservationSystemTestUtil.reservationQ);
+ validateReservation(plan, resID1, reservationDefinition);
+ validateReservation(plan, resID3, reservationDefinition);
+
+ // verify if the reservations have been resized
+ scheduler = rm2.getResourceScheduler();
+ resQ1 = scheduler.getQueueInfo(resID1.toString(), false, false);
+ Assert.assertEquals(1f / 3f, resQ1.getCapacity(), 0.001f);
+ resQ2 = scheduler.getQueueInfo(resID2.toString(), false, false);
+ Assert.assertEquals(1f / 3f, resQ2.getCapacity(), 0.001f);
+ resQ3 = scheduler.getQueueInfo(resID3.toString(), false, false);
+ Assert.assertEquals(1f / 3f, resQ3.getCapacity(), 0.001f);
+ }
+
+ private void waitForReservationActivation(MockRM rm,
+ ReservationId reservationId, String planName) {
+ try {
+ int attempts = 20;
+ do {
+ rm.getRMContext().getReservationSystem().synchronizePlan(planName,
+ false);
+ if (rm.getResourceScheduler()
+ .getQueueInfo(reservationId.toString(), false, false)
+ .getCapacity() > 0f) {
+ break;
+ }
+ LOG.info("Waiting for reservation to be active");
+ Thread.sleep(100);
+ } while (attempts-- > 0);
+ if (attempts <= 0) {
+ Assert
+ .fail("Exceeded attempts in waiting for reservation to be active");
+ }
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/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 7729172..05933f5 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
@@ -215,7 +215,6 @@ public class ReservationSystemTestUtil {
return context;
}
- @SuppressWarnings("unchecked")
public CapacityScheduler mockCapacityScheduler(int numContainers)
throws IOException {
// stolen from TestCapacityScheduler
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestCapacityOverTimePolicy.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/TestCapacityOverTimePolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestCapacityOverTimePolicy.java
index 22ce6aa..4aed064 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestCapacityOverTimePolicy.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestCapacityOverTimePolicy.java
@@ -75,10 +75,11 @@ public class TestCapacityOverTimePolicy {
maxAlloc = Resource.newInstance(1024 * 8, 8);
mAgent = mock(ReservationAgent.class);
- ReservationSystemTestUtil testUtil = new ReservationSystemTestUtil();
QueueMetrics rootQueueMetrics = mock(QueueMetrics.class);
- String reservationQ = testUtil.getFullReservationQueueName();
- Resource clusterResource = testUtil.calculateClusterResource(totCont);
+ String reservationQ =
+ ReservationSystemTestUtil.getFullReservationQueueName();
+ Resource clusterResource =
+ ReservationSystemTestUtil.calculateClusterResource(totCont);
ReservationSchedulerConfiguration conf =
ReservationSystemTestUtil.createConf(reservationQ, timeWindow,
instConstraint, avgConstraint);
@@ -113,7 +114,7 @@ public class TestCapacityOverTimePolicy {
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
"dedicated", initTime, initTime + f.length,
ReservationSystemTestUtil.generateAllocation(initTime, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
}
@Test
@@ -130,7 +131,7 @@ public class TestCapacityOverTimePolicy {
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
"dedicated", initTime, initTime + f.length,
ReservationSystemTestUtil.generateAllocation(initTime, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
}
@Test
@@ -146,7 +147,7 @@ public class TestCapacityOverTimePolicy {
ReservationSystemTestUtil.getNewReservationId(), rDef, "u" + i,
"dedicated", initTime, initTime + f.length,
ReservationSystemTestUtil.generateAllocation(initTime, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
}
}
@@ -163,7 +164,7 @@ public class TestCapacityOverTimePolicy {
ReservationSystemTestUtil.getNewReservationId(), rDef, "u" + i,
"dedicated", initTime, initTime + f.length,
ReservationSystemTestUtil.generateAllocation(initTime, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
}
}
@@ -179,7 +180,7 @@ public class TestCapacityOverTimePolicy {
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
"dedicated", initTime, initTime + f.length,
ReservationSystemTestUtil.generateAllocation(initTime, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
Assert.fail("should not have accepted this");
}
@@ -195,20 +196,20 @@ public class TestCapacityOverTimePolicy {
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
"dedicated", initTime, initTime + f.length,
ReservationSystemTestUtil.generateAllocation(initTime, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
"dedicated", initTime, initTime + f.length,
ReservationSystemTestUtil.generateAllocation(initTime, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
try {
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
"dedicated", initTime, initTime + f.length,
ReservationSystemTestUtil.generateAllocation(initTime, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
Assert.fail();
} catch (PlanningQuotaException p) {
// expected
@@ -232,7 +233,7 @@ public class TestCapacityOverTimePolicy {
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
- "dedicated", initTime, initTime + win, req, res, minAlloc)));
+ "dedicated", initTime, initTime + win, req, res, minAlloc), false));
}
@Test
@@ -251,13 +252,13 @@ public class TestCapacityOverTimePolicy {
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
- "dedicated", initTime, initTime + win, req, res, minAlloc)));
+ "dedicated", initTime, initTime + win, req, res, minAlloc), false));
try {
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(
ReservationSystemTestUtil.getNewReservationId(), null, "u1",
- "dedicated", initTime, initTime + win, req, res, minAlloc)));
+ "dedicated", initTime, initTime + win, req, res, minAlloc), false));
Assert.fail("should not have accepted this");
} catch (PlanningQuotaException e) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestInMemoryPlan.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/TestInMemoryPlan.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestInMemoryPlan.java
index c661271..2e262a0 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestInMemoryPlan.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestInMemoryPlan.java
@@ -113,7 +113,7 @@ public class TestInMemoryPlan {
start, start + alloc.length, allocs, resCalc, minAlloc);
Assert.assertNull(plan.getReservationById(reservationID));
try {
- plan.addReservation(rAllocation);
+ plan.addReservation(rAllocation, false);
} catch (PlanningException e) {
Assert.fail(e.getMessage());
}
@@ -147,7 +147,7 @@ public class TestInMemoryPlan {
start, start + alloc.length, allocs, resCalc, minAlloc);
Assert.assertNull(plan.getReservationById(reservationID));
try {
- plan.addReservation(rAllocation);
+ plan.addReservation(rAllocation, false);
} catch (PlanningException e) {
Assert.fail(e.getMessage());
}
@@ -175,7 +175,7 @@ public class TestInMemoryPlan {
start, start + alloc.length, allocs, resCalc, minAlloc);
Assert.assertNull(plan.getReservationById(reservationID));
try {
- plan.addReservation(rAllocation);
+ plan.addReservation(rAllocation, false);
} catch (PlanningException e) {
Assert.fail(e.getMessage());
}
@@ -189,7 +189,7 @@ public class TestInMemoryPlan {
// Try to add it again
try {
- plan.addReservation(rAllocation);
+ plan.addReservation(rAllocation, false);
Assert.fail("Add should fail as it already exists");
} catch (IllegalArgumentException e) {
Assert.assertTrue(e.getMessage().endsWith("already exists"));
@@ -221,7 +221,7 @@ public class TestInMemoryPlan {
start, start + alloc.length, allocs, resCalc, minAlloc);
Assert.assertNull(plan.getReservationById(reservationID));
try {
- plan.addReservation(rAllocation);
+ plan.addReservation(rAllocation, false);
} catch (PlanningException e) {
Assert.fail(e.getMessage());
}
@@ -316,7 +316,7 @@ public class TestInMemoryPlan {
start, start + alloc.length, allocs, resCalc, minAlloc);
Assert.assertNull(plan.getReservationById(reservationID));
try {
- plan.addReservation(rAllocation);
+ plan.addReservation(rAllocation, false);
} catch (PlanningException e) {
Assert.fail(e.getMessage());
}
@@ -388,7 +388,7 @@ public class TestInMemoryPlan {
minAlloc);
Assert.assertNull(plan.getReservationById(reservationID1));
try {
- plan.addReservation(rAllocation);
+ plan.addReservation(rAllocation, false);
} catch (PlanningException e) {
Assert.fail(e.getMessage());
}
@@ -419,7 +419,7 @@ public class TestInMemoryPlan {
minAlloc);
Assert.assertNull(plan.getReservationById(reservationID2));
try {
- plan.addReservation(rAllocation);
+ plan.addReservation(rAllocation, false);
} catch (PlanningException e) {
Assert.fail(e.getMessage());
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestNoOverCommitPolicy.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/TestNoOverCommitPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestNoOverCommitPolicy.java
index a5f3fb4df..28dd62e 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestNoOverCommitPolicy.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestNoOverCommitPolicy.java
@@ -61,10 +61,11 @@ public class TestNoOverCommitPolicy {
maxAlloc = Resource.newInstance(1024 * 8, 8);
mAgent = mock(ReservationAgent.class);
- ReservationSystemTestUtil testUtil = new ReservationSystemTestUtil();
- String reservationQ = testUtil.getFullReservationQueueName();
+ String reservationQ =
+ ReservationSystemTestUtil.getFullReservationQueueName();
QueueMetrics rootQueueMetrics = mock(QueueMetrics.class);
- Resource clusterResource = testUtil.calculateClusterResource(totCont);
+ Resource clusterResource =
+ ReservationSystemTestUtil.calculateClusterResource(totCont);
ReservationSchedulerConfiguration conf = mock
(ReservationSchedulerConfiguration.class);
NoOverCommitPolicy policy = new NoOverCommitPolicy();
@@ -97,7 +98,7 @@ public class TestNoOverCommitPolicy {
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
"dedicated", initTime, initTime + f.length,
ReservationSystemTestUtil.generateAllocation(initTime, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
}
@Test
@@ -113,7 +114,7 @@ public class TestNoOverCommitPolicy {
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
"dedicated", initTime, initTime + f.length,
ReservationSystemTestUtil.generateAllocation(initTime, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
}
@Test(expected = ResourceOverCommitException.class)
@@ -123,7 +124,7 @@ public class TestNoOverCommitPolicy {
plan.addReservation(new InMemoryReservationAllocation(
ReservationSystemTestUtil.getNewReservationId(), null, "u1",
"dedicated", initTime, initTime + f.length, ReservationSystemTestUtil
- .generateAllocation(initTime, step, f), res, minAlloc));
+ .generateAllocation(initTime, step, f), res, minAlloc), false);
}
@Test(expected = MismatchedUserException.class)
@@ -137,7 +138,7 @@ public class TestNoOverCommitPolicy {
plan.addReservation(new InMemoryReservationAllocation(rid, rDef, "u1",
"dedicated", initTime, initTime + f.length, ReservationSystemTestUtil
- .generateAllocation(initTime, step, f), res, minAlloc));
+ .generateAllocation(initTime, step, f), res, minAlloc), false);
// trying to update a reservation with a mismatching user
plan.updateReservation(new InMemoryReservationAllocation(rid, rDef, "u2",
@@ -158,7 +159,7 @@ public class TestNoOverCommitPolicy {
ReservationSystemTestUtil.getNewReservationId(), rDef, "u" + i,
"dedicated", initTime, initTime + f.length,
ReservationSystemTestUtil.generateAllocation(initTime, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
}
}
@@ -175,7 +176,7 @@ public class TestNoOverCommitPolicy {
ReservationSystemTestUtil.getNewReservationId(), rDef, "u" + i,
"dedicated", initTime, initTime + f.length,
ReservationSystemTestUtil.generateAllocation(initTime, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
}
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestSchedulerPlanFollowerBase.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/TestSchedulerPlanFollowerBase.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestSchedulerPlanFollowerBase.java
index 9689fce..b604799 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestSchedulerPlanFollowerBase.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestSchedulerPlanFollowerBase.java
@@ -18,6 +18,11 @@
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.when;
+
import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
@@ -38,12 +43,6 @@ import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import org.junit.Assert;
-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.when;
-
public abstract class TestSchedulerPlanFollowerBase {
final static int GB = 1024;
protected Clock mClock = null;
@@ -75,20 +74,20 @@ public abstract class TestSchedulerPlanFollowerBase {
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(r1, rDef, "u3",
"dedicated", 0, 0 + f1.length, ReservationSystemTestUtil
- .generateAllocation(0L, 1L, f1), res, minAlloc)));
+ .generateAllocation(0L, 1L, f1), res, minAlloc), false));
ReservationId r2 = ReservationId.newInstance(ts, 2);
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(r2, rDef, "u3",
"dedicated", 3, 3 + f1.length, ReservationSystemTestUtil
- .generateAllocation(3L, 1L, f1), res, minAlloc)));
+ .generateAllocation(3L, 1L, f1), res, minAlloc), false));
ReservationId r3 = ReservationId.newInstance(ts, 3);
int[] f2 = { 0, 10, 20, 10, 0 };
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(r3, rDef, "u4",
"dedicated", 10, 10 + f2.length, ReservationSystemTestUtil
- .generateAllocation(10L, 1L, f2), res, minAlloc)));
+ .generateAllocation(10L, 1L, f2), res, minAlloc), false));
AbstractSchedulerPlanFollower planFollower = createPlanFollower();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/TestAlignedPlanner.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/planning/TestAlignedPlanner.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/TestAlignedPlanner.java
index 01b7976..ec305a2 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/TestAlignedPlanner.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/TestAlignedPlanner.java
@@ -711,9 +711,8 @@ public class TestAlignedPlanner {
Resource clusterCapacity = Resource.newInstance(capacityMem, capacityCores);
- // Set configuration
- ReservationSystemTestUtil testUtil = new ReservationSystemTestUtil();
- String reservationQ = testUtil.getFullReservationQueueName();
+ String reservationQ =
+ ReservationSystemTestUtil.getFullReservationQueueName();
float instConstraint = 100;
float avgConstraint = 100;
@@ -792,7 +791,7 @@ public class TestAlignedPlanner {
ReservationSystemTestUtil.getNewReservationId(), rDef,
"user_fixed", "dedicated", start, start + f.length * step,
ReservationSystemTestUtil.generateAllocation(start, step, f), res,
- minAlloc)));
+ minAlloc), false));
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/TestGreedyReservationAgent.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/planning/TestGreedyReservationAgent.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/TestGreedyReservationAgent.java
index f51cc75..cb4eaeb 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/TestGreedyReservationAgent.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/TestGreedyReservationAgent.java
@@ -21,7 +21,6 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
-
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
@@ -76,8 +75,8 @@ public class TestGreedyReservationAgent {
long timeWindow = 1000000L;
Resource clusterCapacity = Resource.newInstance(100 * 1024, 100);
step = 1000L;
- ReservationSystemTestUtil testUtil = new ReservationSystemTestUtil();
- String reservationQ = testUtil.getFullReservationQueueName();
+ String reservationQ =
+ ReservationSystemTestUtil.getFullReservationQueueName();
float instConstraint = 100;
float avgConstraint = 100;
@@ -151,7 +150,7 @@ public class TestGreedyReservationAgent {
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
"dedicated", 30 * step, 30 * step + f.length * step,
ReservationSystemTestUtil.generateAllocation(30 * step, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
// create a chain of 4 RR, mixing gang and non-gang
ReservationDefinition rr = new ReservationDefinitionPBImpl();
@@ -208,7 +207,7 @@ public class TestGreedyReservationAgent {
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
"dedicated", 30 * step, 30 * step + f.length * step,
ReservationSystemTestUtil.generateAllocation(30 * step, step, f),
- res, minAlloc)));
+ res, minAlloc), false));
// create a chain of 4 RR, mixing gang and non-gang
ReservationDefinition rr = new ReservationDefinitionPBImpl();
@@ -529,7 +528,7 @@ public class TestGreedyReservationAgent {
plan.addReservation(new InMemoryReservationAllocation(
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
"dedicated", 0L, 0L + f.length * step, ReservationSystemTestUtil
- .generateAllocation(0, step, f), res, minAlloc)));
+ .generateAllocation(0, step, f), res, minAlloc), false));
int[] f2 = { 5, 5, 5, 5, 5, 5, 5 };
Map<ReservationInterval, Resource> alloc =
@@ -537,7 +536,8 @@ public class TestGreedyReservationAgent {
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
- "dedicated", 5000, 5000 + f2.length * step, alloc, res, minAlloc)));
+ "dedicated", 5000, 5000 + f2.length * step, alloc, res, minAlloc),
+ false));
System.out.println("--------BEFORE AGENT----------");
System.out.println(plan.toString());
@@ -563,7 +563,8 @@ public class TestGreedyReservationAgent {
step = 1000L;
ReservationSystemTestUtil testUtil = new ReservationSystemTestUtil();
CapacityScheduler scheduler = testUtil.mockCapacityScheduler(500 * 100);
- String reservationQ = testUtil.getFullReservationQueueName();
+ String reservationQ =
+ ReservationSystemTestUtil.getFullReservationQueueName();
float instConstraint = 100;
float avgConstraint = 100;
ReservationSchedulerConfiguration conf =
http://git-wip-us.apache.org/repos/asf/hadoop/blob/2798723a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/TestSimpleCapacityReplanner.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/planning/TestSimpleCapacityReplanner.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/TestSimpleCapacityReplanner.java
index b6e6667..eb0b0e2 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/TestSimpleCapacityReplanner.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/planning/TestSimpleCapacityReplanner.java
@@ -93,44 +93,44 @@ public class TestSimpleCapacityReplanner {
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(r1, rDef, "u3",
"dedicated", 0, 0 + f5.length, generateAllocation(0, f5), res,
- minAlloc)));
+ minAlloc), false));
when(clock.getTime()).thenReturn(1L);
ReservationId r2 = ReservationId.newInstance(ts, 2);
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(r2, rDef, "u4",
"dedicated", 0, 0 + f5.length, generateAllocation(0, f5), res,
- minAlloc)));
+ minAlloc), false));
when(clock.getTime()).thenReturn(2L);
ReservationId r3 = ReservationId.newInstance(ts, 3);
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(r3, rDef, "u5",
"dedicated", 0, 0 + f5.length, generateAllocation(0, f5), res,
- minAlloc)));
+ minAlloc), false));
when(clock.getTime()).thenReturn(3L);
ReservationId r4 = ReservationId.newInstance(ts, 4);
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(r4, rDef, "u6",
"dedicated", 0, 0 + f5.length, generateAllocation(0, f5), res,
- minAlloc)));
+ minAlloc), false));
when(clock.getTime()).thenReturn(4L);
ReservationId r5 = ReservationId.newInstance(ts, 5);
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(r5, rDef, "u7",
"dedicated", 0, 0 + f5.length, generateAllocation(0, f5), res,
- minAlloc)));
+ minAlloc), false));
int[] f6 = { 50, 50, 50, 50, 50 };
ReservationId r6 = ReservationId.newInstance(ts, 6);
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(r6, rDef, "u3",
"dedicated", 10, 10 + f6.length, generateAllocation(10, f6), res,
- minAlloc)));
+ minAlloc), false));
when(clock.getTime()).thenReturn(6L);
ReservationId r7 = ReservationId.newInstance(ts, 7);
assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation(r7, rDef, "u4",
"dedicated", 10, 10 + f6.length, generateAllocation(10, f6), res,
- minAlloc)));
+ minAlloc), false));
// remove some of the resources (requires replanning)
plan.setTotalCapacity(Resource.newInstance(70 * 1024, 70));