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/05/26 21:02:17 UTC
hadoop git commit: YARN-3632. Ordering policy should be allowed to
reorder an application when demand changes. Contributed by Craig Welch
Repository: hadoop
Updated Branches:
refs/heads/trunk 500a1d9c7 -> 10732d515
YARN-3632. Ordering policy should be allowed to reorder an application when demand changes. Contributed by Craig Welch
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/10732d51
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/10732d51
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/10732d51
Branch: refs/heads/trunk
Commit: 10732d515f62258309f98e4d7d23249f80b1847d
Parents: 500a1d9
Author: Jian He <ji...@apache.org>
Authored: Tue May 26 12:00:51 2015 -0700
Committer: Jian He <ji...@apache.org>
Committed: Tue May 26 12:00:51 2015 -0700
----------------------------------------------------------------------
hadoop-yarn-project/CHANGES.txt | 3 +
.../scheduler/AppSchedulingInfo.java | 7 +-
.../scheduler/SchedulerApplicationAttempt.java | 5 +-
.../scheduler/capacity/CapacityScheduler.java | 19 ++-
.../AbstractComparatorOrderingPolicy.java | 25 ++++
.../scheduler/policy/FairOrderingPolicy.java | 11 +-
.../scheduler/policy/FifoOrderingPolicy.java | 6 +-
.../scheduler/policy/OrderingPolicy.java | 5 +
.../capacity/TestCapacityScheduler.java | 115 +++++++++++++++++++
9 files changed, 187 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/10732d51/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index 481c5a9..6cdd9f3 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -443,6 +443,9 @@ Release 2.8.0 - UNRELEASED
YARN-3707. RM Web UI queue filter doesn't work. (Wangda Tan via jianhe)
+ YARN-3632. Ordering policy should be allowed to reorder an application when
+ demand changes. (Craig Welch via jianhe)
+
Release 2.7.1 - UNRELEASED
INCOMPATIBLE CHANGES
http://git-wip-us.apache.org/repos/asf/hadoop/blob/10732d51/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.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/AppSchedulingInfo.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
index 5604f0f..77ac5b3 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
@@ -128,11 +128,14 @@ public class AppSchedulingInfo {
*
* @param requests resources to be acquired
* @param recoverPreemptedRequest recover Resource Request on preemption
+ * @return true if any resource was updated, false else
*/
- synchronized public void updateResourceRequests(
+ synchronized public boolean updateResourceRequests(
List<ResourceRequest> requests, boolean recoverPreemptedRequest) {
QueueMetrics metrics = queue.getMetrics();
+ boolean anyResourcesUpdated = false;
+
// Update resource requests
for (ResourceRequest request : requests) {
Priority priority = request.getPriority();
@@ -146,6 +149,7 @@ public class AppSchedulingInfo {
+ request);
}
updatePendingResources = true;
+ anyResourcesUpdated = true;
// Premature optimization?
// Assumes that we won't see more than one priority request updated
@@ -209,6 +213,7 @@ public class AppSchedulingInfo {
}
}
}
+ return anyResourcesUpdated;
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/10732d51/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.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/SchedulerApplicationAttempt.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
index 0554c04..dbc3cb5 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
@@ -284,11 +284,12 @@ public class SchedulerApplicationAttempt implements SchedulableEntity {
return queue;
}
- public synchronized void updateResourceRequests(
+ public synchronized boolean updateResourceRequests(
List<ResourceRequest> requests) {
if (!isStopped) {
- appSchedulingInfo.updateResourceRequests(requests, false);
+ return appSchedulingInfo.updateResourceRequests(requests, false);
}
+ return false;
}
public synchronized void recoverResourceRequests(
http://git-wip-us.apache.org/repos/asf/hadoop/blob/10732d51/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 48c7f2f..06d282d 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
@@ -895,6 +895,10 @@ public class CapacityScheduler extends
// Release containers
releaseContainers(release, application);
+ Allocation allocation;
+
+ LeafQueue updateDemandForQueue = null;
+
synchronized (application) {
// make sure we aren't stopping/removing the application
@@ -915,8 +919,10 @@ public class CapacityScheduler extends
application.showRequests();
// Update application requests
- application.updateResourceRequests(ask);
-
+ if (application.updateResourceRequests(ask)) {
+ updateDemandForQueue = (LeafQueue) application.getQueue();
+ }
+
LOG.debug("allocate: post-update");
application.showRequests();
}
@@ -929,9 +935,16 @@ public class CapacityScheduler extends
application.updateBlacklist(blacklistAdditions, blacklistRemovals);
- return application.getAllocation(getResourceCalculator(),
+ allocation = application.getAllocation(getResourceCalculator(),
clusterResource, getMinimumResourceCapability());
}
+
+ if (updateDemandForQueue != null) {
+ updateDemandForQueue.getOrderingPolicy().demandUpdated(application);
+ }
+
+ return allocation;
+
}
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/10732d51/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/AbstractComparatorOrderingPolicy.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/policy/AbstractComparatorOrderingPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/AbstractComparatorOrderingPolicy.java
index e046fcf..c4d2aae 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/AbstractComparatorOrderingPolicy.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/AbstractComparatorOrderingPolicy.java
@@ -37,6 +37,7 @@ public abstract class AbstractComparatorOrderingPolicy<S extends SchedulableEnti
protected TreeSet<S> schedulableEntities;
protected Comparator<SchedulableEntity> comparator;
+ protected Map<String, S> entitiesToReorder = new HashMap<String, S>();
public AbstractComparatorOrderingPolicy() { }
@@ -47,11 +48,13 @@ public abstract class AbstractComparatorOrderingPolicy<S extends SchedulableEnti
@Override
public Iterator<S> getAssignmentIterator() {
+ reorderScheduleEntities();
return schedulableEntities.iterator();
}
@Override
public Iterator<S> getPreemptionIterator() {
+ reorderScheduleEntities();
return schedulableEntities.descendingIterator();
}
@@ -68,6 +71,22 @@ public abstract class AbstractComparatorOrderingPolicy<S extends SchedulableEnti
schedulableEntities.add(schedulableEntity);
}
+ protected void reorderScheduleEntities() {
+ synchronized (entitiesToReorder) {
+ for (Map.Entry<String, S> entry :
+ entitiesToReorder.entrySet()) {
+ reorderSchedulableEntity(entry.getValue());
+ }
+ entitiesToReorder.clear();
+ }
+ }
+
+ protected void entityRequiresReordering(S schedulableEntity) {
+ synchronized (entitiesToReorder) {
+ entitiesToReorder.put(schedulableEntity.getId(), schedulableEntity);
+ }
+ }
+
@VisibleForTesting
public Comparator<SchedulableEntity> getComparator() {
return comparator;
@@ -80,6 +99,9 @@ public abstract class AbstractComparatorOrderingPolicy<S extends SchedulableEnti
@Override
public boolean removeSchedulableEntity(S s) {
+ synchronized (entitiesToReorder) {
+ entitiesToReorder.remove(s.getId());
+ }
return schedulableEntities.remove(s);
}
@@ -105,6 +127,9 @@ public abstract class AbstractComparatorOrderingPolicy<S extends SchedulableEnti
RMContainer r);
@Override
+ public abstract void demandUpdated(S schedulableEntity);
+
+ @Override
public abstract String getInfo();
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/10732d51/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/FairOrderingPolicy.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/policy/FairOrderingPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/FairOrderingPolicy.java
index 3ab74de..ea14b42 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/FairOrderingPolicy.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/FairOrderingPolicy.java
@@ -96,16 +96,23 @@ public class FairOrderingPolicy<S extends SchedulableEntity> extends AbstractCom
@Override
public void containerAllocated(S schedulableEntity,
RMContainer r) {
- reorderSchedulableEntity(schedulableEntity);
+ entityRequiresReordering(schedulableEntity);
}
@Override
public void containerReleased(S schedulableEntity,
RMContainer r) {
- reorderSchedulableEntity(schedulableEntity);
+ entityRequiresReordering(schedulableEntity);
}
@Override
+ public void demandUpdated(S schedulableEntity) {
+ if (sizeBasedWeight) {
+ entityRequiresReordering(schedulableEntity);
+ }
+ }
+
+ @Override
public String getInfo() {
String sbw = sizeBasedWeight ? " with sizeBasedWeight" : "";
return "FairOrderingPolicy" + sbw;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/10732d51/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/FifoOrderingPolicy.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/policy/FifoOrderingPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/FifoOrderingPolicy.java
index 932a5f9..74a422c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/FifoOrderingPolicy.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/FifoOrderingPolicy.java
@@ -46,7 +46,11 @@ public class FifoOrderingPolicy<S extends SchedulableEntity> extends AbstractCom
public void containerReleased(S schedulableEntity,
RMContainer r) {
}
-
+
+ @Override
+ public void demandUpdated(S schedulableEntity) {
+ }
+
@Override
public String getInfo() {
return "FifoOrderingPolicy";
http://git-wip-us.apache.org/repos/asf/hadoop/blob/10732d51/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/OrderingPolicy.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/policy/OrderingPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/OrderingPolicy.java
index aebdcde..e3f67ce 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/OrderingPolicy.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/policy/OrderingPolicy.java
@@ -102,6 +102,11 @@ public interface OrderingPolicy<S extends SchedulableEntity> {
RMContainer r);
/**
+ * Demand Updated for the passed schedulableEntity, reorder if needed.
+ */
+ void demandUpdated(S schedulableEntity);
+
+ /**
* Display information regarding configuration & status
*/
public String getInfo();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/10732d51/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java
index d360581..7b665e4 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java
@@ -72,6 +72,8 @@ import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
+import org.apache.hadoop.yarn.factories.RecordFactory;
+import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.nodelabels.RMNodeLabel;
import org.apache.hadoop.yarn.server.api.protocolrecords.UpdateNodeResourceRequest;
@@ -126,6 +128,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.CapacitySchedule
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.CapacitySchedulerLeafQueueInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.CapacitySchedulerQueueInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.CapacitySchedulerQueueInfoList;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.FairOrderingPolicy;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.util.resource.DominantResourceCalculator;
import org.apache.hadoop.yarn.util.resource.Resources;
@@ -677,6 +680,118 @@ public class TestCapacityScheduler {
}
@Test
+ public void testAllocateReorder() throws Exception {
+
+ //Confirm that allocation (resource request) alone will trigger a change in
+ //application ordering where appropriate
+
+ Configuration conf = new Configuration();
+ conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
+ ResourceScheduler.class);
+ MockRM rm = new MockRM(conf);
+ rm.start();
+ CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
+
+ LeafQueue q = (LeafQueue) cs.getQueue("default");
+ Assert.assertNotNull(q);
+
+ FairOrderingPolicy fop = new FairOrderingPolicy();
+ fop.setSizeBasedWeight(true);
+ q.setOrderingPolicy(fop);
+
+ String host = "127.0.0.1";
+ RMNode node =
+ MockNodes.newNodeInfo(0, MockNodes.newResource(4 * GB), 1, host);
+ cs.handle(new NodeAddedSchedulerEvent(node));
+
+ //add app begin
+ ApplicationId appId1 = BuilderUtils.newApplicationId(100, 1);
+ ApplicationAttemptId appAttemptId1 = BuilderUtils.newApplicationAttemptId(
+ appId1, 1);
+
+ RMAppAttemptMetrics attemptMetric1 =
+ new RMAppAttemptMetrics(appAttemptId1, rm.getRMContext());
+ RMAppImpl app1 = mock(RMAppImpl.class);
+ when(app1.getApplicationId()).thenReturn(appId1);
+ RMAppAttemptImpl attempt1 = mock(RMAppAttemptImpl.class);
+ when(attempt1.getAppAttemptId()).thenReturn(appAttemptId1);
+ when(attempt1.getRMAppAttemptMetrics()).thenReturn(attemptMetric1);
+ when(app1.getCurrentAppAttempt()).thenReturn(attempt1);
+
+ rm.getRMContext().getRMApps().put(appId1, app1);
+
+ SchedulerEvent addAppEvent1 =
+ new AppAddedSchedulerEvent(appId1, "default", "user");
+ cs.handle(addAppEvent1);
+ SchedulerEvent addAttemptEvent1 =
+ new AppAttemptAddedSchedulerEvent(appAttemptId1, false);
+ cs.handle(addAttemptEvent1);
+ //add app end
+
+ //add app begin
+ ApplicationId appId2 = BuilderUtils.newApplicationId(100, 2);
+ ApplicationAttemptId appAttemptId2 = BuilderUtils.newApplicationAttemptId(
+ appId2, 1);
+
+ RMAppAttemptMetrics attemptMetric2 =
+ new RMAppAttemptMetrics(appAttemptId2, rm.getRMContext());
+ RMAppImpl app2 = mock(RMAppImpl.class);
+ when(app2.getApplicationId()).thenReturn(appId2);
+ RMAppAttemptImpl attempt2 = mock(RMAppAttemptImpl.class);
+ when(attempt2.getAppAttemptId()).thenReturn(appAttemptId2);
+ when(attempt2.getRMAppAttemptMetrics()).thenReturn(attemptMetric2);
+ when(app2.getCurrentAppAttempt()).thenReturn(attempt2);
+
+ rm.getRMContext().getRMApps().put(appId2, app2);
+
+ SchedulerEvent addAppEvent2 =
+ new AppAddedSchedulerEvent(appId2, "default", "user");
+ cs.handle(addAppEvent2);
+ SchedulerEvent addAttemptEvent2 =
+ new AppAttemptAddedSchedulerEvent(appAttemptId2, false);
+ cs.handle(addAttemptEvent2);
+ //add app end
+
+ RecordFactory recordFactory =
+ RecordFactoryProvider.getRecordFactory(null);
+
+ Priority priority = TestUtils.createMockPriority(1);
+ ResourceRequest r1 = TestUtils.createResourceRequest(ResourceRequest.ANY, 1*GB, 1, true, priority, recordFactory);
+
+ //This will allocate for app1
+ cs.allocate(appAttemptId1,
+ Collections.<ResourceRequest>singletonList(r1),
+ Collections.<ContainerId>emptyList(),
+ null, null);
+
+ //And this will result in container assignment for app1
+ CapacityScheduler.schedule(cs);
+
+ //Verify that app1 is still first in assignment order
+ //This happens because app2 has no demand/a magnitude of NaN, which
+ //results in app1 and app2 being equal in the fairness comparison and
+ //failling back to fifo (start) ordering
+ assertEquals(q.getOrderingPolicy().getAssignmentIterator().next().getId(),
+ appId1.toString());
+
+ //Now, allocate for app2 (this would be the first/AM allocation)
+ ResourceRequest r2 = TestUtils.createResourceRequest(ResourceRequest.ANY, 1*GB, 1, true, priority, recordFactory);
+ cs.allocate(appAttemptId2,
+ Collections.<ResourceRequest>singletonList(r2),
+ Collections.<ContainerId>emptyList(),
+ null, null);
+
+ //In this case we do not perform container assignment because we want to
+ //verify re-ordering based on the allocation alone
+
+ //Now, the first app for assignment is app2
+ assertEquals(q.getOrderingPolicy().getAssignmentIterator().next().getId(),
+ appId2.toString());
+
+ rm.stop();
+ }
+
+ @Test
public void testResourceOverCommit() throws Exception {
Configuration conf = new Configuration();
conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,