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 ae...@apache.org on 2017/03/08 23:46:00 UTC
[02/33] hadoop git commit: YARN-6218. Fix TestAMRMClient when using
FairScheduler. (Miklos Szegedi via rchiang)
YARN-6218. Fix TestAMRMClient when using FairScheduler. (Miklos Szegedi via rchiang)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/97398708
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/97398708
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/97398708
Branch: refs/heads/HDFS-7240
Commit: 973987089090b428ae34a86926c8ef8ebca45aa5
Parents: 4a09ed0
Author: Ray Chiang <rc...@apache.org>
Authored: Fri Mar 3 12:55:45 2017 -0800
Committer: Anu Engineer <ae...@apache.org>
Committed: Wed Mar 8 15:33:55 2017 -0800
----------------------------------------------------------------------
.../yarn/client/api/impl/TestAMRMClient.java | 139 ++++++++++++-------
.../scheduler/fair/FairScheduler.java | 3 +-
2 files changed, 87 insertions(+), 55 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/97398708/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java
index 4f73bac..43c0271 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java
@@ -64,47 +64,71 @@ import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
import org.apache.hadoop.yarn.server.MiniYARNCluster;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
+import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.security.AMRMTokenSecretManager;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.util.Records;
import org.junit.After;
-import org.junit.AfterClass;
import org.junit.Assert;
+import org.junit.Assume;
import org.junit.Before;
-import org.junit.BeforeClass;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.eclipse.jetty.util.log.Log;
import com.google.common.base.Supplier;
+/**
+ * Test application master client class to resource manager.
+ */
+@RunWith(value = Parameterized.class)
public class TestAMRMClient {
- static Configuration conf = null;
- static MiniYARNCluster yarnCluster = null;
- static YarnClient yarnClient = null;
- static List<NodeReport> nodeReports = null;
- static ApplicationAttemptId attemptId = null;
- static int nodeCount = 3;
+ private String schedulerName = null;
+ private Configuration conf = null;
+ private MiniYARNCluster yarnCluster = null;
+ private YarnClient yarnClient = null;
+ private List<NodeReport> nodeReports = null;
+ private ApplicationAttemptId attemptId = null;
+ private int nodeCount = 3;
static final int rolling_interval_sec = 13;
static final long am_expire_ms = 4000;
- static Resource capability;
- static Priority priority;
- static Priority priority2;
- static String node;
- static String rack;
- static String[] nodes;
- static String[] racks;
+ private Resource capability;
+ private Priority priority;
+ private Priority priority2;
+ private String node;
+ private String rack;
+ private String[] nodes;
+ private String[] racks;
private final static int DEFAULT_ITERATION = 3;
- @BeforeClass
- public static void setup() throws Exception {
+ public TestAMRMClient(String schedulerName) {
+ this.schedulerName = schedulerName;
+ }
+
+ @Parameterized.Parameters
+ public static Collection<Object[]> data() {
+ List<Object[]> list = new ArrayList<Object[]>(2);
+ list.add(new Object[] {CapacityScheduler.class.getName()});
+ list.add(new Object[] {FairScheduler.class.getName()});
+ return list;
+ }
+
+ @Before
+ public void setup() throws Exception {
// start minicluster
conf = new YarnConfiguration();
+ conf.set(YarnConfiguration.RM_SCHEDULER, schedulerName);
conf.setLong(
YarnConfiguration.RM_AMRM_TOKEN_MASTER_KEY_ROLLING_INTERVAL_SECS,
rolling_interval_sec);
@@ -138,10 +162,7 @@ public class TestAMRMClient {
rack = nodeReports.get(0).getRackName();
nodes = new String[]{ node };
racks = new String[]{ rack };
- }
-
- @Before
- public void startApp() throws Exception {
+
// submit new app
ApplicationSubmissionContext appContext =
yarnClient.createApplication().getApplicationSubmissionContext();
@@ -199,13 +220,10 @@ public class TestAMRMClient {
}
@After
- public void cancelApp() throws YarnException, IOException {
+ public void teardown() throws YarnException, IOException {
yarnClient.killApplication(attemptId.getApplicationId());
attemptId = null;
- }
-
- @AfterClass
- public static void tearDown() {
+
if (yarnClient != null && yarnClient.getServiceState() == STATE.STARTED) {
yarnClient.stop();
}
@@ -656,8 +674,8 @@ public class TestAMRMClient {
amClient.releaseAssignedContainer(container.getId());
}
if(allocatedContainerCount < containersRequestedAny) {
- // sleep to let NM's heartbeat to RM and trigger allocations
- sleep(100);
+ // let NM heartbeat to RM and trigger allocations
+ triggerSchedulingWithNMHeartBeat();
}
}
@@ -678,7 +696,27 @@ public class TestAMRMClient {
}
}
}
-
+
+ /**
+ * Make sure we get allocations regardless of timing issues.
+ */
+ private void triggerSchedulingWithNMHeartBeat() {
+ // Simulate fair scheduler update thread
+ RMContext context = yarnCluster.getResourceManager().getRMContext();
+ if (context.getScheduler() instanceof FairScheduler) {
+ FairScheduler scheduler = (FairScheduler)context.getScheduler();
+ scheduler.update();
+ }
+ // Trigger NM's heartbeat to RM and trigger allocations
+ for (RMNode rmNode : context.getRMNodes().values()) {
+ context.getScheduler().handle(new NodeUpdateSchedulerEvent(rmNode));
+ }
+ if (context.getScheduler() instanceof FairScheduler) {
+ FairScheduler scheduler = (FairScheduler)context.getScheduler();
+ scheduler.update();
+ }
+ }
+
@Test (timeout=60000)
public void testAllocationWithBlacklist() throws YarnException, IOException {
AMRMClientImpl<ContainerRequest> amClient = null;
@@ -811,8 +849,8 @@ public class TestAMRMClient {
allocatedContainerCount += allocResponse.getAllocatedContainers().size();
if(allocatedContainerCount == 0) {
- // sleep to let NM's heartbeat to RM and trigger allocations
- sleep(100);
+ // let NM heartbeat to RM and trigger allocations
+ triggerSchedulingWithNMHeartBeat();
}
}
return allocatedContainerCount;
@@ -934,6 +972,8 @@ public class TestAMRMClient {
@Test(timeout=60000)
public void testAMRMClientWithContainerResourceChange()
throws YarnException, IOException {
+ // Fair scheduler does not support resource change
+ Assume.assumeTrue(schedulerName.equals(CapacityScheduler.class.getName()));
AMRMClient<ContainerRequest> amClient = null;
try {
// start am rm client
@@ -981,8 +1021,8 @@ public class TestAMRMClient {
}
// send allocation requests
amClient.allocate(0.1f);
- // sleep to let NM's heartbeat to RM and trigger allocations
- sleep(150);
+ // let NM heartbeat to RM and trigger allocations
+ triggerSchedulingWithNMHeartBeat();
// get allocations
AllocateResponse allocResponse = amClient.allocate(0.1f);
List<Container> containers = allocResponse.getAllocatedContainers();
@@ -1012,14 +1052,14 @@ public class TestAMRMClient {
if (status.getState() == ContainerState.RUNNING) {
break;
}
- sleep(100);
+ sleep(10);
}
}
} catch (YarnException e) {
throw new AssertionError("Exception is not expected: " + e);
}
- // sleep to let NM's heartbeat to RM to confirm container launch
- sleep(200);
+ // let NM's heartbeat to RM to confirm container launch
+ triggerSchedulingWithNMHeartBeat();
return containers;
}
@@ -1079,7 +1119,7 @@ public class TestAMRMClient {
allocResponse.getUpdatedContainers();
Assert.assertEquals(1, updatedContainers.size());
// we should get increase allocation after the next NM's heartbeat to RM
- sleep(150);
+ triggerSchedulingWithNMHeartBeat();
// get allocations
allocResponse = amClient.allocate(0.1f);
updatedContainers =
@@ -1142,8 +1182,8 @@ public class TestAMRMClient {
}
if(allocatedContainerCount < containersRequestedAny) {
- // sleep to let NM's heartbeat to RM and trigger allocations
- sleep(100);
+ // let NM heartbeat to RM and trigger allocations
+ triggerSchedulingWithNMHeartBeat();
}
}
@@ -1225,8 +1265,8 @@ public class TestAMRMClient {
}
}
if(numIterations > 0) {
- // sleep to make sure NM's heartbeat
- sleep(100);
+ // let NM heartbeat to RM and trigger allocations
+ triggerSchedulingWithNMHeartBeat();
}
}
assertEquals(0, amClient.ask.size());
@@ -1284,8 +1324,8 @@ public class TestAMRMClient {
}
if(allocatedContainers.size() < containersRequestedAny) {
- // sleep to let NM's heartbeat to RM and trigger allocations
- sleep(100);
+ // let NM heartbeat to RM and trigger allocations
+ triggerSchedulingWithNMHeartBeat();
}
}
@@ -1396,12 +1436,7 @@ public class TestAMRMClient {
while (System.currentTimeMillis() - startTime <
rolling_interval_sec * 1000) {
amClient.allocate(0.1f);
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
+ sleep(1000);
}
amClient.allocate(0.1f);
@@ -1461,11 +1496,7 @@ public class TestAMRMClient {
}
}
amClient.allocate(0.1f);
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- // DO NOTHING
- }
+ sleep(1000);
}
try {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/97398708/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
index 0599414..4f3e4f9 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
@@ -349,7 +349,8 @@ public class FairScheduler extends
* fair shares, deficits, minimum slot allocations, and amount of used and
* required resources per job.
*/
- protected void update() {
+ @VisibleForTesting
+ public void update() {
try {
writeLock.lock();
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org