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 zh...@apache.org on 2015/08/11 19:44:53 UTC
[03/21] hadoop git commit: YARN-3887. Support changing Application
priority during runtime. Contributed by Sunil G
YARN-3887. Support changing Application priority during runtime. Contributed by Sunil G
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/fa1d84ae
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/fa1d84ae
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/fa1d84ae
Branch: refs/heads/HDFS-7285-merge
Commit: fa1d84ae2739a1e76f58b9c96d1378f9453cc0d2
Parents: b56daff
Author: Jian He <ji...@apache.org>
Authored: Mon Aug 10 20:51:54 2015 -0700
Committer: Jian He <ji...@apache.org>
Committed: Mon Aug 10 20:51:54 2015 -0700
----------------------------------------------------------------------
hadoop-yarn-project/CHANGES.txt | 3 +
.../resourcemanager/recovery/RMStateStore.java | 5 +
.../scheduler/AbstractYarnScheduler.java | 7 +
.../scheduler/SchedulerApplicationAttempt.java | 2 +-
.../scheduler/YarnScheduler.java | 11 +
.../scheduler/capacity/CapacityScheduler.java | 49 ++++
.../AbstractComparatorOrderingPolicy.java | 6 +
.../capacity/TestApplicationPriority.java | 260 +++++++++++++++++++
8 files changed, 342 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa1d84ae/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index 5e27a2f..ada1056 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -164,6 +164,9 @@ Release 2.8.0 - UNRELEASED
YARN-3873. PendingApplications in LeafQueue should also use OrderingPolicy.
(Sunil G via wangda)
+ YARN-3887. Support changing Application priority during runtime. (Sunil G
+ via jianhe)
+
IMPROVEMENTS
YARN-644. Basic null check is not performed on passed in arguments before
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa1d84ae/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.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/recovery/RMStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java
index 5036450..affbee1 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java
@@ -706,6 +706,11 @@ public abstract class RMStateStore extends AbstractService {
dispatcher.getEventHandler().handle(new RMStateUpdateAppEvent(appState));
}
+ public void updateApplicationStateSynchronously(
+ ApplicationStateData appState) {
+ handleStoreEvent(new RMStateUpdateAppEvent(appState));
+ }
+
public void updateFencedState() {
handleStoreEvent(new RMStateStoreEvent(RMStateStoreEventType.FENCED));
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa1d84ae/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java
index d69600a..ed05189 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractYarnScheduler.java
@@ -701,4 +701,11 @@ public abstract class AbstractYarnScheduler
// specific scheduler.
return Priority.newInstance(0);
}
+
+ @Override
+ public void updateApplicationPriority(Priority newPriority,
+ ApplicationId applicationId) throws YarnException {
+ // Dummy Implementation till Application Priority changes are done in
+ // specific scheduler.
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa1d84ae/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 317e61c..4872543 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
@@ -98,7 +98,7 @@ public class SchedulerApplicationAttempt implements SchedulableEntity {
private boolean amRunning = false;
private LogAggregationContext logAggregationContext;
- private Priority appPriority = null;
+ private volatile Priority appPriority = null;
protected ResourceUsage attemptResourceUsage = new ResourceUsage();
private AtomicLong firstAllocationRequestSentTime = new AtomicLong(0);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa1d84ae/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/YarnScheduler.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/YarnScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/YarnScheduler.java
index f629579..0fa23e1 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/YarnScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/YarnScheduler.java
@@ -306,4 +306,15 @@ public interface YarnScheduler extends EventHandler<SchedulerEvent> {
public Priority checkAndGetApplicationPriority(Priority priorityFromContext,
String user, String queueName, ApplicationId applicationId)
throws YarnException;
+
+ /**
+ *
+ * Change application priority of a submitted application at runtime
+ *
+ * @param newPriority Submitted Application priority.
+ *
+ * @param applicationId Application ID
+ */
+ public void updateApplicationPriority(Priority newPriority,
+ ApplicationId applicationId) throws YarnException;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa1d84ae/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 b4d0095..b4b1383 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
@@ -71,6 +71,7 @@ import org.apache.hadoop.yarn.security.YarnAuthorizationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationConstants;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
@@ -1850,4 +1851,52 @@ public class CapacityScheduler extends
public Priority getMaxClusterLevelAppPriority() {
return maxClusterLevelAppPriority;
}
+
+ @Override
+ public synchronized void updateApplicationPriority(Priority newPriority,
+ ApplicationId applicationId) throws YarnException {
+ Priority appPriority = null;
+ SchedulerApplication<FiCaSchedulerApp> application = applications
+ .get(applicationId);
+
+ if (application == null) {
+ throw new YarnException("Application '" + applicationId
+ + "' is not present, hence could not change priority.");
+ }
+
+ if (application.getPriority().equals(newPriority)) {
+ return;
+ }
+
+ RMApp rmApp = rmContext.getRMApps().get(applicationId);
+ appPriority = checkAndGetApplicationPriority(newPriority, rmApp.getUser(),
+ rmApp.getQueue(), applicationId);
+
+ // Update new priority in Submission Context to keep track in HA
+ rmApp.getApplicationSubmissionContext().setPriority(appPriority);
+
+ // Update to state store
+ ApplicationStateData appState = ApplicationStateData.newInstance(
+ rmApp.getSubmitTime(), rmApp.getStartTime(),
+ rmApp.getApplicationSubmissionContext(), rmApp.getUser());
+ rmContext.getStateStore().updateApplicationStateSynchronously(appState);
+
+ // As we use iterator over a TreeSet for OrderingPolicy, once we change
+ // priority then reinsert back to make order correct.
+ LeafQueue queue = (LeafQueue) getQueue(rmApp.getQueue());
+ synchronized (queue) {
+ queue.getOrderingPolicy().removeSchedulableEntity(
+ application.getCurrentAppAttempt());
+
+ // Update new priority in SchedulerApplication
+ application.setPriority(appPriority);
+
+ queue.getOrderingPolicy().addSchedulableEntity(
+ application.getCurrentAppAttempt());
+ }
+
+ LOG.info("Priority '" + appPriority + "' is updated in queue :"
+ + rmApp.getQueue() + "for application:" + applicationId
+ + "for the user: " + rmApp.getUser());
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa1d84ae/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 c4d2aae..7bec03a 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
@@ -94,11 +94,17 @@ public abstract class AbstractComparatorOrderingPolicy<S extends SchedulableEnti
@Override
public void addSchedulableEntity(S s) {
+ if (null == s) {
+ return;
+ }
schedulableEntities.add(s);
}
@Override
public boolean removeSchedulableEntity(S s) {
+ if (null == s) {
+ return false;
+ }
synchronized (entitiesToReorder) {
entitiesToReorder.remove(s.getId());
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/fa1d84ae/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestApplicationPriority.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/TestApplicationPriority.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/TestApplicationPriority.java
index db094e3..169e9f6 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/TestApplicationPriority.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/TestApplicationPriority.java
@@ -23,7 +23,9 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
@@ -36,6 +38,9 @@ import org.apache.hadoop.yarn.server.resourcemanager.MockAM;
import org.apache.hadoop.yarn.server.resourcemanager.MockNM;
import org.apache.hadoop.yarn.server.resourcemanager.MockNodes;
import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.MemoryRMStateStore;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppImpl;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptImpl;
@@ -307,4 +312,259 @@ public class TestApplicationPriority {
maxPriority);
rm.stop();
}
+
+ @Test
+ public void testUpdatePriorityAtRuntime() throws Exception {
+
+ Configuration conf = new Configuration();
+ conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
+ ResourceScheduler.class);
+ // Set Max Application Priority as 10
+ conf.setInt(YarnConfiguration.MAX_CLUSTER_LEVEL_APPLICATION_PRIORITY, 10);
+ MockRM rm = new MockRM(conf);
+ rm.start();
+
+ Priority appPriority1 = Priority.newInstance(5);
+ MockNM nm1 = rm.registerNode("127.0.0.1:1234", 16 * GB);
+ RMApp app1 = rm.submitApp(1 * GB, appPriority1);
+
+ // kick the scheduler, 1 GB given to AM1, remaining 15GB on nm1
+ MockAM am1 = MockRM.launchAM(app1, rm, nm1);
+ am1.registerAppAttempt();
+
+ // get scheduler
+ CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
+
+ // Change the priority of App1 to 8
+ Priority appPriority2 = Priority.newInstance(8);
+ cs.updateApplicationPriority(appPriority2, app1.getApplicationId());
+
+ // get scheduler app
+ FiCaSchedulerApp schedulerAppAttempt = cs.getSchedulerApplications()
+ .get(app1.getApplicationId()).getCurrentAppAttempt();
+
+ // Verify whether the new priority is updated
+ Assert.assertEquals(appPriority2, schedulerAppAttempt.getPriority());
+ }
+
+ @Test
+ public void testUpdateInvalidPriorityAtRuntime() throws Exception {
+
+ Configuration conf = new Configuration();
+ conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
+ ResourceScheduler.class);
+ // Set Max Application Priority as 10
+ conf.setInt(YarnConfiguration.MAX_CLUSTER_LEVEL_APPLICATION_PRIORITY, 10);
+ MockRM rm = new MockRM(conf);
+ rm.start();
+
+ Priority appPriority1 = Priority.newInstance(5);
+ MockNM nm1 = rm.registerNode("127.0.0.1:1234", 16 * GB);
+ RMApp app1 = rm.submitApp(1 * GB, appPriority1);
+
+ // kick the scheduler, 1 GB given to AM1, remaining 15GB on nm1
+ MockAM am1 = MockRM.launchAM(app1, rm, nm1);
+ am1.registerAppAttempt();
+
+ // get scheduler
+ CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
+
+ // Change the priority of App1 to 15
+ Priority appPriority2 = Priority.newInstance(15);
+ cs.updateApplicationPriority(appPriority2, app1.getApplicationId());
+
+ // get scheduler app
+ FiCaSchedulerApp schedulerAppAttempt = cs.getSchedulerApplications()
+ .get(app1.getApplicationId()).getCurrentAppAttempt();
+
+ // Verify whether priority 15 is reset to 10
+ Priority appPriority3 = Priority.newInstance(10);
+ Assert.assertEquals(appPriority3, schedulerAppAttempt.getPriority());
+ rm.stop();
+ }
+
+ @Test(timeout = 180000)
+ public void testRMRestartWithChangeInPriority() throws Exception {
+ conf.setBoolean(YarnConfiguration.RECOVERY_ENABLED, true);
+ conf.setBoolean(YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_ENABLED,
+ false);
+ conf.set(YarnConfiguration.RM_STORE, MemoryRMStateStore.class.getName());
+ conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS,
+ YarnConfiguration.DEFAULT_RM_AM_MAX_ATTEMPTS);
+ conf.setInt(YarnConfiguration.MAX_CLUSTER_LEVEL_APPLICATION_PRIORITY, 10);
+
+ MemoryRMStateStore memStore = new MemoryRMStateStore();
+ memStore.init(conf);
+ RMState rmState = memStore.getState();
+ Map<ApplicationId, ApplicationStateData> rmAppState = rmState
+ .getApplicationState();
+
+ // PHASE 1: create state in an RM
+
+ // start RM
+ MockRM rm1 = new MockRM(conf, memStore);
+ rm1.start();
+
+ MockNM nm1 = new MockNM("127.0.0.1:1234", 15120,
+ rm1.getResourceTrackerService());
+ nm1.registerNode();
+
+ Priority appPriority1 = Priority.newInstance(5);
+ RMApp app1 = rm1.submitApp(1 * GB, appPriority1);
+
+ // kick the scheduler, 1 GB given to AM1, remaining 15GB on nm1
+ MockAM am1 = MockRM.launchAM(app1, rm1, nm1);
+ am1.registerAppAttempt();
+
+ // get scheduler
+ CapacityScheduler cs = (CapacityScheduler) rm1.getResourceScheduler();
+
+ // Change the priority of App1 to 8
+ Priority appPriority2 = Priority.newInstance(8);
+ cs.updateApplicationPriority(appPriority2, app1.getApplicationId());
+
+ // let things settle down
+ Thread.sleep(1000);
+
+ // create new RM to represent restart and recover state
+ MockRM rm2 = new MockRM(conf, memStore);
+
+ // start new RM
+ rm2.start();
+ // change NM to point to new RM
+ nm1.setResourceTrackerService(rm2.getResourceTrackerService());
+
+ // Verify RM Apps after this restart
+ Assert.assertEquals(1, rm2.getRMContext().getRMApps().size());
+
+ // get scheduler app
+ RMApp loadedApp = rm2.getRMContext().getRMApps()
+ .get(app1.getApplicationId());
+
+ // Verify whether priority 15 is reset to 10
+ Assert.assertEquals(appPriority2, loadedApp.getCurrentAppAttempt()
+ .getSubmissionContext().getPriority());
+
+ rm2.stop();
+ rm1.stop();
+ }
+
+ @Test
+ public void testApplicationPriorityAllocationWithChangeInPriority()
+ throws Exception {
+
+ Configuration conf = new Configuration();
+ conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
+ ResourceScheduler.class);
+ // Set Max Application Priority as 10
+ conf.setInt(YarnConfiguration.MAX_CLUSTER_LEVEL_APPLICATION_PRIORITY, 10);
+ MockRM rm = new MockRM(conf);
+ rm.start();
+
+ Priority appPriority1 = Priority.newInstance(5);
+ MockNM nm1 = rm.registerNode("127.0.0.1:1234", 16 * GB);
+ RMApp app1 = rm.submitApp(1 * GB, appPriority1);
+
+ // kick the scheduler, 1 GB given to AM1, remaining 15GB on nm1
+ MockAM am1 = MockRM.launchAM(app1, rm, nm1);
+ am1.registerAppAttempt();
+
+ // add request for containers and wait for containers to be allocated.
+ int NUM_CONTAINERS = 7;
+ List<Container> allocated1 = am1.allocateAndWaitForContainers("127.0.0.1",
+ NUM_CONTAINERS, 2 * GB, nm1);
+
+ Assert.assertEquals(7, allocated1.size());
+ Assert.assertEquals(2 * GB, allocated1.get(0).getResource().getMemory());
+
+ // check node report, 15 GB used (1 AM and 7 containers) and 1 GB available
+ SchedulerNodeReport report_nm1 = rm.getResourceScheduler().getNodeReport(
+ nm1.getNodeId());
+ Assert.assertEquals(15 * GB, report_nm1.getUsedResource().getMemory());
+ Assert.assertEquals(1 * GB, report_nm1.getAvailableResource().getMemory());
+
+ // Submit the second app App2 with priority 8 (Higher than App1)
+ Priority appPriority2 = Priority.newInstance(8);
+ RMApp app2 = rm.submitApp(1 * GB, appPriority2);
+
+ // kick the scheduler, 1 GB which was free is given to AM of App2
+ nm1.nodeHeartbeat(true);
+ MockAM am2 = MockRM.launchAM(app2, rm, nm1);
+ am2.registerAppAttempt();
+
+ // check node report, 16 GB used and 0 GB available
+ report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId());
+ Assert.assertEquals(16 * GB, report_nm1.getUsedResource().getMemory());
+ Assert.assertEquals(0 * GB, report_nm1.getAvailableResource().getMemory());
+
+ // get scheduler
+ CapacityScheduler cs = (CapacityScheduler) rm.getResourceScheduler();
+
+ // get scheduler app
+ FiCaSchedulerApp schedulerAppAttemptApp1 = cs.getSchedulerApplications()
+ .get(app1.getApplicationId()).getCurrentAppAttempt();
+ // kill 2 containers to free up some space
+ int counter = 0;
+ for (Iterator<Container> iterator = allocated1.iterator(); iterator
+ .hasNext();) {
+ Container c = iterator.next();
+ if (++counter > 2) {
+ break;
+ }
+ cs.killContainer(schedulerAppAttemptApp1.getRMContainer(c.getId()));
+ iterator.remove();
+ }
+
+ // check node report, 12 GB used and 4 GB available
+ report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId());
+ Assert.assertEquals(12 * GB, report_nm1.getUsedResource().getMemory());
+ Assert.assertEquals(4 * GB, report_nm1.getAvailableResource().getMemory());
+
+ // add request for containers App1
+ am1.allocate("127.0.0.1", 2 * GB, 10, new ArrayList<ContainerId>());
+
+ // add request for containers App2 and wait for containers to get allocated
+ List<Container> allocated2 = am2.allocateAndWaitForContainers("127.0.0.1",
+ 2, 2 * GB, nm1);
+
+ Assert.assertEquals(2, allocated2.size());
+ // check node report, 16 GB used and 0 GB available
+ report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId());
+ Assert.assertEquals(16 * GB, report_nm1.getUsedResource().getMemory());
+ Assert.assertEquals(0 * GB, report_nm1.getAvailableResource().getMemory());
+
+ // kill 1 more
+ counter = 0;
+ for (Iterator<Container> iterator = allocated1.iterator(); iterator
+ .hasNext();) {
+ Container c = iterator.next();
+ if (++counter > 1) {
+ break;
+ }
+ cs.killContainer(schedulerAppAttemptApp1.getRMContainer(c.getId()));
+ iterator.remove();
+ }
+
+ // check node report, 14 GB used and 2 GB available
+ report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId());
+ Assert.assertEquals(14 * GB, report_nm1.getUsedResource().getMemory());
+ Assert.assertEquals(2 * GB, report_nm1.getAvailableResource().getMemory());
+
+ // Change the priority of App1 to 3 (lowest)
+ Priority appPriority3 = Priority.newInstance(3);
+ cs.updateApplicationPriority(appPriority3, app2.getApplicationId());
+
+ // add request for containers App2
+ am2.allocate("127.0.0.1", 2 * GB, 3, new ArrayList<ContainerId>());
+
+ // add request for containers App1 and wait for containers to get allocated
+ // since priority is more for App1 now, App1 will get a container.
+ List<Container> allocated3 = am1.allocateAndWaitForContainers("127.0.0.1",
+ 1, 2 * GB, nm1);
+
+ Assert.assertEquals(1, allocated3.size());
+ // Now App1 will have 5 containers and 1 AM. App2 will have 2 containers.
+ Assert.assertEquals(6, schedulerAppAttemptApp1.getLiveContainers().size());
+ rm.stop();
+ }
}