Skip to content

Commit eec611e

Browse files
author
Devarajulu K
committed
YARN-879. Fixed tests w.r.t o.a.h.y.server.resourcemanager.Application. Contributed by Junping Du.
git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1530902 13f79535-47bb-0310-9956-ffa450edef68
1 parent d284ef7 commit eec611e

File tree

6 files changed

+127
-73
lines changed

6 files changed

+127
-73
lines changed

‎hadoop-yarn-project/CHANGES.txt‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,9 @@ Release 2.2.1 - UNRELEASED
9595
so that clients don't need to do scheme-mangling. (Omkar Vinit Joshi via
9696
vinodkv)
9797

98+
YARN-879. Fixed tests w.r.t o.a.h.y.server.resourcemanager.Application.
99+
(Junping Du via devaraj)
100+
98101
Release 2.2.0 - 2013-10-13
99102

100103
INCOMPATIBLE CHANGES

‎hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/Application.java‎

Lines changed: 41 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,8 @@
3434
import org.apache.commons.logging.LogFactory;
3535
import org.apache.hadoop.classification.InterfaceAudience.Private;
3636
import org.apache.hadoop.net.NetworkTopology;
37+
import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest;
38+
import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse;
3739
import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
3840
import org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest;
3941
import org.apache.hadoop.yarn.api.protocolrecords.StopContainersRequest;
@@ -47,11 +49,16 @@
4749
import org.apache.hadoop.yarn.api.records.Priority;
4850
import org.apache.hadoop.yarn.api.records.Resource;
4951
import org.apache.hadoop.yarn.api.records.ResourceRequest;
52+
import org.apache.hadoop.yarn.conf.YarnConfiguration;
5053
import org.apache.hadoop.yarn.exceptions.YarnException;
5154
import org.apache.hadoop.yarn.factories.RecordFactory;
5255
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
5356
import org.apache.hadoop.yarn.server.resourcemanager.Task.State;
57+
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
5458
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
59+
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
60+
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppAddedSchedulerEvent;
61+
import org.apache.hadoop.yarn.util.Records;
5562
import org.apache.hadoop.yarn.util.resource.Resources;
5663

5764
@Private
@@ -89,16 +96,23 @@ public class Application {
8996

9097
Resource used = recordFactory.newRecordInstance(Resource.class);
9198

92-
public Application(String user, ResourceManager resourceManager) {
99+
public Application(String user, ResourceManager resourceManager)
100+
throws YarnException {
93101
this(user, "default", resourceManager);
94102
}
95103

96-
public Application(String user, String queue, ResourceManager resourceManager) {
104+
public Application(String user, String queue, ResourceManager resourceManager)
105+
throws YarnException {
97106
this.user = user;
98107
this.queue = queue;
99108
this.resourceManager = resourceManager;
100-
this.applicationId =
101-
this.resourceManager.getClientRMService().getNewApplicationId();
109+
// register an application
110+
GetNewApplicationRequest request =
111+
Records.newRecord(GetNewApplicationRequest.class);
112+
GetNewApplicationResponse newApp =
113+
this.resourceManager.getClientRMService().getNewApplication(request);
114+
this.applicationId = newApp.getApplicationId();
115+
102116
this.applicationAttemptId =
103117
ApplicationAttemptId.newInstance(this.applicationId,
104118
this.numAttempts.getAndIncrement());
@@ -115,6 +129,10 @@ public String getQueue() {
115129
public ApplicationId getApplicationId() {
116130
return applicationId;
117131
}
132+
133+
public ApplicationAttemptId getApplicationAttemptId() {
134+
return applicationAttemptId;
135+
}
118136

119137
public static String resolve(String hostName) {
120138
return NetworkTopology.DEFAULT_RACK;
@@ -132,10 +150,25 @@ public synchronized void submit() throws IOException, YarnException {
132150
ApplicationSubmissionContext context = recordFactory.newRecordInstance(ApplicationSubmissionContext.class);
133151
context.setApplicationId(this.applicationId);
134152
context.setQueue(this.queue);
153+
154+
// Set up the container launch context for the application master
155+
ContainerLaunchContext amContainer
156+
= Records.newRecord(ContainerLaunchContext.class);
157+
context.setAMContainerSpec(amContainer);
158+
context.setResource(Resources.createResource(
159+
YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB));
160+
135161
SubmitApplicationRequest request = recordFactory
136162
.newRecordInstance(SubmitApplicationRequest.class);
137163
request.setApplicationSubmissionContext(context);
164+
final ResourceScheduler scheduler = resourceManager.getResourceScheduler();
165+
138166
resourceManager.getClientRMService().submitApplication(request);
167+
168+
// Notify scheduler
169+
AppAddedSchedulerEvent appAddedEvent1 = new AppAddedSchedulerEvent(
170+
this.applicationAttemptId, this.queue, this.user);
171+
scheduler.handle(appAddedEvent1);
139172
}
140173

141174
public synchronized void addResourceRequestSpec(
@@ -267,17 +300,13 @@ public synchronized List<Container> getResources() throws IOException {
267300
}
268301

269302
// Get resources from the ResourceManager
270-
resourceManager.getResourceScheduler().allocate(applicationAttemptId,
271-
new ArrayList<ResourceRequest>(ask), new ArrayList<ContainerId>(), null, null);
303+
Allocation allocation = resourceManager.getResourceScheduler().allocate(
304+
applicationAttemptId, new ArrayList<ResourceRequest>(ask),
305+
new ArrayList<ContainerId>(), null, null);
272306
System.out.println("-=======" + applicationAttemptId);
273307
System.out.println("----------" + resourceManager.getRMContext().getRMApps()
274308
.get(applicationId).getRMAppAttempt(applicationAttemptId));
275-
276-
List<Container> containers = null;
277-
// TODO: Fix
278-
// resourceManager.getRMContext().getRMApps()
279-
// .get(applicationId).getRMAppAttempt(applicationAttemptId)
280-
// .pullNewlyAllocatedContainers();
309+
List<Container> containers = allocation.getContainers();
281310

282311
// Clear state for next interaction with ResourceManager
283312
ask.clear();

‎hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/NodeManager.java‎

Lines changed: 12 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,6 @@
5656
import org.apache.hadoop.yarn.server.api.protocolrecords.RegisterNodeManagerRequest;
5757
import org.apache.hadoop.yarn.server.api.records.NodeHealthStatus;
5858
import org.apache.hadoop.yarn.server.api.records.NodeStatus;
59-
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode;
6059
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
6160
import org.apache.hadoop.yarn.util.YarnVersionInfo;
6261
import org.apache.hadoop.yarn.util.resource.Resources;
@@ -71,28 +70,27 @@ public class NodeManager implements ContainerManagementProtocol {
7170
final private String rackName;
7271
final private NodeId nodeId;
7372
final private Resource capability;
73+
final private ResourceManager resourceManager;
7474
Resource available = recordFactory.newRecordInstance(Resource.class);
7575
Resource used = recordFactory.newRecordInstance(Resource.class);
7676

7777
final ResourceTrackerService resourceTrackerService;
78-
final FiCaSchedulerNode schedulerNode;
7978
final Map<ApplicationId, List<Container>> containers =
8079
new HashMap<ApplicationId, List<Container>>();
8180

8281
final Map<Container, ContainerStatus> containerStatusMap =
8382
new HashMap<Container, ContainerStatus>();
84-
83+
8584
public NodeManager(String hostName, int containerManagerPort, int httpPort,
8685
String rackName, Resource capability,
87-
ResourceTrackerService resourceTrackerService, RMContext rmContext)
86+
ResourceManager resourceManager)
8887
throws IOException, YarnException {
8988
this.containerManagerAddress = hostName + ":" + containerManagerPort;
9089
this.nodeHttpAddress = hostName + ":" + httpPort;
9190
this.rackName = rackName;
92-
this.resourceTrackerService = resourceTrackerService;
91+
this.resourceTrackerService = resourceManager.getResourceTrackerService();
9392
this.capability = capability;
9493
Resources.addTo(available, capability);
95-
9694
this.nodeId = NodeId.newInstance(hostName, containerManagerPort);
9795
RegisterNodeManagerRequest request = recordFactory
9896
.newRecordInstance(RegisterNodeManagerRequest.class);
@@ -101,14 +99,8 @@ public NodeManager(String hostName, int containerManagerPort, int httpPort,
10199
request.setNodeId(this.nodeId);
102100
request.setNMVersion(YarnVersionInfo.getVersion());
103101
resourceTrackerService.registerNodeManager(request);
104-
this.schedulerNode = new FiCaSchedulerNode(rmContext.getRMNodes().get(
105-
this.nodeId), false);
106-
107-
// Sanity check
108-
Assert.assertEquals(capability.getMemory(),
109-
schedulerNode.getAvailableResource().getMemory());
110-
Assert.assertEquals(capability.getVirtualCores(),
111-
schedulerNode.getAvailableResource().getVirtualCores());
102+
this.resourceManager = resourceManager;
103+
resourceManager.getResourceScheduler().getNodeReport(this.nodeId);
112104
}
113105

114106
public String getHostName() {
@@ -220,9 +212,11 @@ synchronized public StartContainersResponse startContainers(
220212
synchronized public void checkResourceUsage() {
221213
LOG.info("Checking resource usage for " + containerManagerAddress);
222214
Assert.assertEquals(available.getMemory(),
223-
schedulerNode.getAvailableResource().getMemory());
215+
resourceManager.getResourceScheduler().getNodeReport(
216+
this.nodeId).getAvailableResource().getMemory());
224217
Assert.assertEquals(used.getMemory(),
225-
schedulerNode.getUsedResource().getMemory());
218+
resourceManager.getResourceScheduler().getNodeReport(
219+
this.nodeId).getUsedResource().getMemory());
226220
}
227221

228222
@Override
@@ -232,9 +226,9 @@ synchronized public StopContainersResponse stopContainers(StopContainersRequest
232226
String applicationId =
233227
String.valueOf(containerID.getApplicationAttemptId()
234228
.getApplicationId().getId());
235-
236229
// Mark the container as COMPLETE
237-
List<Container> applicationContainers = containers.get(applicationId);
230+
List<Container> applicationContainers = containers.get(containerID.getApplicationAttemptId()
231+
.getApplicationId());
238232
for (Container c : applicationContainers) {
239233
if (c.getId().compareTo(containerID) == 0) {
240234
ContainerStatus containerStatus = containerStatusMap.get(c);

‎hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestResourceManager.java‎

Lines changed: 34 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,11 @@
3434
import org.apache.hadoop.yarn.conf.YarnConfiguration;
3535
import org.apache.hadoop.yarn.exceptions.YarnException;
3636
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
37+
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
3738
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
39+
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppRemovedSchedulerEvent;
40+
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeAddedSchedulerEvent;
41+
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent;
3842
import org.apache.hadoop.yarn.util.resource.Resources;
3943
import org.junit.After;
4044
import org.junit.Before;
@@ -62,13 +66,18 @@ public void tearDown() throws Exception {
6266
registerNode(String hostName, int containerManagerPort, int httpPort,
6367
String rackName, Resource capability) throws IOException,
6468
YarnException {
65-
return new org.apache.hadoop.yarn.server.resourcemanager.NodeManager(
66-
hostName, containerManagerPort, httpPort, rackName, capability,
67-
resourceManager.getResourceTrackerService(), resourceManager
68-
.getRMContext());
69+
org.apache.hadoop.yarn.server.resourcemanager.NodeManager nm =
70+
new org.apache.hadoop.yarn.server.resourcemanager.NodeManager(
71+
hostName, containerManagerPort, httpPort, rackName, capability,
72+
resourceManager);
73+
NodeAddedSchedulerEvent nodeAddEvent1 =
74+
new NodeAddedSchedulerEvent(resourceManager.getRMContext()
75+
.getRMNodes().get(nm.getNodeId()));
76+
resourceManager.getResourceScheduler().handle(nodeAddEvent1);
77+
return nm;
6978
}
7079

71-
// @Test
80+
@Test
7281
public void testResourceAllocation() throws IOException,
7382
YarnException {
7483
LOG.info("--- START: testResourceAllocation ---");
@@ -80,14 +89,12 @@ public void testResourceAllocation() throws IOException,
8089
org.apache.hadoop.yarn.server.resourcemanager.NodeManager nm1 =
8190
registerNode(host1, 1234, 2345, NetworkTopology.DEFAULT_RACK,
8291
Resources.createResource(memory, 1));
83-
nm1.heartbeat();
8492

8593
// Register node2
8694
String host2 = "host2";
8795
org.apache.hadoop.yarn.server.resourcemanager.NodeManager nm2 =
8896
registerNode(host2, 1234, 2345, NetworkTopology.DEFAULT_RACK,
8997
Resources.createResource(memory/2, 1));
90-
nm2.heartbeat();
9198

9299
// Submit an application
93100
Application application = new Application("user1", resourceManager);
@@ -105,23 +112,22 @@ public void testResourceAllocation() throws IOException,
105112

106113
Task t1 = new Task(application, priority1, new String[] {host1, host2});
107114
application.addTask(t1);
108-
115+
109116
final int memory2 = 2048;
110117
Resource capability2 = Resources.createResource(memory2, 1);
111118
Priority priority0 =
112-
org.apache.hadoop.yarn.server.resourcemanager.resource.Priority.create(0); // higher
119+
org.apache.hadoop.yarn.server.resourcemanager.resource.Priority.create(0); // higher
113120
application.addResourceRequestSpec(priority0, capability2);
114121

115122
// Send resource requests to the scheduler
116123
application.schedule();
117-
118-
// Send a heartbeat to kick the tires on the Scheduler
119-
nm1.heartbeat();
124+
125+
// Send a heartbeat to kick the tires on the Scheduler
126+
nodeUpdate(nm1);
120127

121128
// Get allocations from the scheduler
122129
application.schedule();
123130

124-
nm1.heartbeat();
125131
checkResourceUsage(nm1, nm2);
126132

127133
LOG.info("Adding new tasks...");
@@ -137,18 +143,13 @@ public void testResourceAllocation() throws IOException,
137143
checkResourceUsage(nm1, nm2);
138144

139145
// Send a heartbeat to kick the tires on the Scheduler
140-
LOG.info("Sending hb from host2");
141-
nm2.heartbeat();
142-
143-
LOG.info("Sending hb from host1");
144-
nm1.heartbeat();
146+
nodeUpdate(nm2);
147+
nodeUpdate(nm1);
145148

146149
// Get allocations from the scheduler
147150
LOG.info("Trying to allocate...");
148151
application.schedule();
149152

150-
nm1.heartbeat();
151-
nm2.heartbeat();
152153
checkResourceUsage(nm1, nm2);
153154

154155
// Complete tasks
@@ -157,13 +158,23 @@ public void testResourceAllocation() throws IOException,
157158
application.finishTask(t2);
158159
application.finishTask(t3);
159160

160-
// Send heartbeat
161-
nm1.heartbeat();
162-
nm2.heartbeat();
161+
// Notify scheduler application is finished.
162+
AppRemovedSchedulerEvent appRemovedEvent1 = new AppRemovedSchedulerEvent(
163+
application.getApplicationAttemptId(), RMAppAttemptState.FINISHED);
164+
resourceManager.getResourceScheduler().handle(appRemovedEvent1);
165+
163166
checkResourceUsage(nm1, nm2);
164167

165168
LOG.info("--- END: testResourceAllocation ---");
166169
}
170+
171+
private void nodeUpdate(
172+
org.apache.hadoop.yarn.server.resourcemanager.NodeManager nm1) {
173+
RMNode node = resourceManager.getRMContext().getRMNodes().get(nm1.getNodeId());
174+
// Send a heartbeat to kick the tires on the Scheduler
175+
NodeUpdateSchedulerEvent nodeUpdate = new NodeUpdateSchedulerEvent(node);
176+
resourceManager.getResourceScheduler().handle(nodeUpdate);
177+
}
167178

168179
@Test
169180
public void testNodeHealthReportIsNotNull() throws Exception{

0 commit comments

Comments
 (0)