Skip to content

Commit 19c743c

Browse files
wangdatanwangdatan
committed
YARN-4889. Changes in AMRMClient for identifying resource-requests explicitly. (Arun Suresh via wangda)
1 parent b930dc3 commit 19c743c

File tree

9 files changed

+468
-112
lines changed

9 files changed

+468
-112
lines changed

‎hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,10 @@ public int compare(ResourceRequest r1, ResourceRequest r2) {
111111

112112
// Compare priority, host and capability
113113
int ret = r1.getPriority().compareTo(r2.getPriority());
114+
if (ret == 0) {
115+
ret = Long.compare(
116+
r1.getAllocationRequestId(), r2.getAllocationRequestId());
117+
}
114118
if (ret == 0) {
115119
String h1 = r1.getResourceName();
116120
String h2 = r2.getResourceName();
@@ -381,6 +385,7 @@ public int hashCode() {
381385
result = prime * result + ((hostName == null) ? 0 : hostName.hashCode());
382386
result = prime * result + getNumContainers();
383387
result = prime * result + ((priority == null) ? 0 : priority.hashCode());
388+
result = prime * result + Long.valueOf(getAllocationRequestId()).hashCode();
384389
return result;
385390
}
386391

@@ -422,6 +427,11 @@ public boolean equals(Object obj) {
422427
.equals(other.getExecutionTypeRequest().getExecutionType())) {
423428
return false;
424429
}
430+
431+
if (getAllocationRequestId() != other.getAllocationRequestId()) {
432+
return false;
433+
}
434+
425435
if (getNodeLabelExpression() == null) {
426436
if (other.getNodeLabelExpression() != null) {
427437
return false;
@@ -452,7 +462,14 @@ public int compareTo(ResourceRequest other) {
452462
int capabilityComparison =
453463
this.getCapability().compareTo(other.getCapability());
454464
if (capabilityComparison == 0) {
455-
return this.getNumContainers() - other.getNumContainers();
465+
int numContainerComparison =
466+
this.getNumContainers() - other.getNumContainers();
467+
if (numContainerComparison == 0) {
468+
return Long.compare(getAllocationRequestId(),
469+
other.getAllocationRequestId());
470+
} else {
471+
return numContainerComparison;
472+
}
456473
} else {
457474
return capabilityComparison;
458475
}

‎hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java‎

Lines changed: 128 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,7 @@ public static class ContainerRequest {
110110
final List<String> nodes;
111111
final List<String> racks;
112112
final Priority priority;
113+
final long allocationRequestId;
113114
final boolean relaxLocality;
114115
final String nodeLabelsExpression;
115116
final ExecutionTypeRequest executionTypeRequest;
@@ -134,6 +135,31 @@ public ContainerRequest(Resource capability, String[] nodes,
134135
String[] racks, Priority priority) {
135136
this(capability, nodes, racks, priority, true, null);
136137
}
138+
139+
/**
140+
* Instantiates a {@link ContainerRequest} with the given constraints and
141+
* locality relaxation enabled.
142+
*
143+
* @param capability
144+
* The {@link Resource} to be requested for each container.
145+
* @param nodes
146+
* Any hosts to request that the containers are placed on.
147+
* @param racks
148+
* Any racks to request that the containers are placed on. The
149+
* racks corresponding to any hosts requested will be automatically
150+
* added to this list.
151+
* @param priority
152+
* The priority at which to request the containers. Higher
153+
* priorities have lower numerical values.
154+
* @param allocationRequestId Allocation Request Id
155+
*/
156+
@Public
157+
@InterfaceStability.Evolving
158+
public ContainerRequest(Resource capability, String[] nodes,
159+
String[] racks, Priority priority, long allocationRequestId) {
160+
this(capability, nodes, racks, priority, allocationRequestId, true, null,
161+
ExecutionTypeRequest.newInstance());
162+
}
137163

138164
/**
139165
* Instantiates a {@link ContainerRequest} with the given constraints.
@@ -158,6 +184,34 @@ public ContainerRequest(Resource capability, String[] nodes,
158184
this(capability, nodes, racks, priority, relaxLocality, null);
159185
}
160186

187+
/**
188+
* Instantiates a {@link ContainerRequest} with the given constraints.
189+
*
190+
* @param capability
191+
* The {@link Resource} to be requested for each container.
192+
* @param nodes
193+
* Any hosts to request that the containers are placed on.
194+
* @param racks
195+
* Any racks to request that the containers are placed on. The
196+
* racks corresponding to any hosts requested will be automatically
197+
* added to this list.
198+
* @param priority
199+
* The priority at which to request the containers. Higher
200+
* priorities have lower numerical values.
201+
* @param relaxLocality
202+
* If true, containers for this request may be assigned on hosts
203+
* and racks other than the ones explicitly requested.
204+
* @param allocationRequestId Allocation Request Id
205+
*/
206+
@Public
207+
@InterfaceStability.Evolving
208+
public ContainerRequest(Resource capability, String[] nodes,
209+
String[] racks, Priority priority, long allocationRequestId,
210+
boolean relaxLocality) {
211+
this(capability, nodes, racks, priority, allocationRequestId,
212+
relaxLocality, null, ExecutionTypeRequest.newInstance());
213+
}
214+
161215
/**
162216
* Instantiates a {@link ContainerRequest} with the given constraints.
163217
*
@@ -181,10 +235,45 @@ public ContainerRequest(Resource capability, String[] nodes,
181235
*/
182236
public ContainerRequest(Resource capability, String[] nodes, String[] racks,
183237
Priority priority, boolean relaxLocality, String nodeLabelsExpression) {
184-
this(capability, nodes, racks, priority, relaxLocality,
238+
this(capability, nodes, racks, priority, 0, relaxLocality,
185239
nodeLabelsExpression,
186240
ExecutionTypeRequest.newInstance());
187241
}
242+
243+
/**
244+
* Instantiates a {@link ContainerRequest} with the given constraints.
245+
*
246+
* @param capability
247+
* The {@link Resource} to be requested for each container.
248+
* @param nodes
249+
* Any hosts to request that the containers are placed on.
250+
* @param racks
251+
* Any racks to request that the containers are placed on. The
252+
* racks corresponding to any hosts requested will be automatically
253+
* added to this list.
254+
* @param priority
255+
* The priority at which to request the containers. Higher
256+
* priorities have lower numerical values.
257+
* @param allocationRequestId
258+
* The allocationRequestId of the request. To be used as a tracking
259+
* id to match Containers allocated against this request. Will
260+
* default to 0 if not specified.
261+
* @param relaxLocality
262+
* If true, containers for this request may be assigned on hosts
263+
* and racks other than the ones explicitly requested.
264+
* @param nodeLabelsExpression
265+
* Set node labels to allocate resource, now we only support
266+
* asking for only a single node label
267+
*/
268+
@Public
269+
@InterfaceStability.Evolving
270+
public ContainerRequest(Resource capability, String[] nodes, String[] racks,
271+
Priority priority, long allocationRequestId, boolean relaxLocality,
272+
String nodeLabelsExpression) {
273+
this(capability, nodes, racks, priority, allocationRequestId,
274+
relaxLocality, nodeLabelsExpression,
275+
ExecutionTypeRequest.newInstance());
276+
}
188277

189278
/**
190279
* Instantiates a {@link ContainerRequest} with the given constraints.
@@ -200,6 +289,10 @@ public ContainerRequest(Resource capability, String[] nodes, String[] racks,
200289
* @param priority
201290
* The priority at which to request the containers. Higher
202291
* priorities have lower numerical values.
292+
* @param allocationRequestId
293+
* The allocationRequestId of the request. To be used as a tracking
294+
* id to match Containers allocated against this request. Will
295+
* default to 0 if not specified.
203296
* @param relaxLocality
204297
* If true, containers for this request may be assigned on hosts
205298
* and racks other than the ones explicitly requested.
@@ -210,7 +303,8 @@ public ContainerRequest(Resource capability, String[] nodes, String[] racks,
210303
* Set the execution type of the container request.
211304
*/
212305
public ContainerRequest(Resource capability, String[] nodes, String[] racks,
213-
Priority priority, boolean relaxLocality, String nodeLabelsExpression,
306+
Priority priority, long allocationRequestId, boolean relaxLocality,
307+
String nodeLabelsExpression,
214308
ExecutionTypeRequest executionTypeRequest) {
215309
// Validate request
216310
Preconditions.checkArgument(capability != null,
@@ -223,6 +317,7 @@ public ContainerRequest(Resource capability, String[] nodes, String[] racks,
223317
&& (nodes == null || nodes.length == 0)),
224318
"Can't turn off locality relaxation on a " +
225319
"request with no location constraints");
320+
this.allocationRequestId = allocationRequestId;
226321
this.capability = capability;
227322
this.nodes = (nodes != null ? ImmutableList.copyOf(nodes) : null);
228323
this.racks = (racks != null ? ImmutableList.copyOf(racks) : null);
@@ -247,6 +342,10 @@ public List<String> getRacks() {
247342
public Priority getPriority() {
248343
return priority;
249344
}
345+
346+
public long getAllocationRequestId() {
347+
return allocationRequestId;
348+
}
250349

251350
public boolean getRelaxLocality() {
252351
return relaxLocality;
@@ -264,6 +363,7 @@ public String toString() {
264363
StringBuilder sb = new StringBuilder();
265364
sb.append("Capability[").append(capability).append("]");
266365
sb.append("Priority[").append(priority).append("]");
366+
sb.append("AllocationRequestId[").append(allocationRequestId).append("]");
267367
sb.append("ExecutionTypeRequest[").append(executionTypeRequest)
268368
.append("]");
269369
return sb.toString();
@@ -390,6 +490,10 @@ public abstract void requestContainerResourceChange(
390490
* Each collection in the list contains requests with identical
391491
* <code>Resource</code> size that fit in the given capability. In a
392492
* collection, requests will be returned in the same order as they were added.
493+
*
494+
* NOTE: This API only matches Container requests that were created by the
495+
* client WITHOUT the allocationRequestId being set.
496+
*
393497
* @return Collection of request matching the parameters
394498
*/
395499
@InterfaceStability.Evolving
@@ -407,7 +511,11 @@ public abstract List<? extends Collection<T>> getMatchingRequests(
407511
* Each collection in the list contains requests with identical
408512
* <code>Resource</code> size that fit in the given capability. In a
409513
* collection, requests will be returned in the same order as they were added.
410-
* specify an <code>ExecutionType</code> .
514+
* specify an <code>ExecutionType</code>.
515+
*
516+
* NOTE: This API only matches Container requests that were created by the
517+
* client WITHOUT the allocationRequestId being set.
518+
*
411519
* @param priority Priority
412520
* @param resourceName Location
413521
* @param executionType ExecutionType
@@ -421,7 +529,23 @@ public List<? extends Collection<T>> getMatchingRequests(
421529
throw new UnsupportedOperationException("The sub-class extending" +
422530
" AMRMClient is expected to implement this !!");
423531
}
424-
532+
533+
/**
534+
* Get outstanding <code>ContainerRequest</code>s matching the given
535+
* allocationRequestId. These ContainerRequests should have been added via
536+
* <code>addContainerRequest</code> earlier in the lifecycle. For performance,
537+
* the AMRMClient may return its internal collection directly without creating
538+
* a copy. Users should not perform mutable operations on the return value.
539+
*
540+
* NOTE: This API only matches Container requests that were created by the
541+
* client WITH the allocationRequestId being set to a non-default value.
542+
*
543+
* @param allocationRequestId Allocation Request Id
544+
* @return Collection of request matching the parameters
545+
*/
546+
@InterfaceStability.Evolving
547+
public abstract Collection<T> getMatchingRequests(long allocationRequestId);
548+
425549
/**
426550
* Update application's blacklist with addition or removal resources.
427551
*

‎hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -202,6 +202,10 @@ public abstract List<? extends Collection<T>> getMatchingRequests(
202202
/**
203203
* Returns all matching ContainerRequests that match the given Priority,
204204
* ResourceName, ExecutionType and Capability.
205+
*
206+
* NOTE: This matches only requests that were made by the client WITHOUT the
207+
* allocationRequestId specified.
208+
*
205209
* @param priority Priority.
206210
* @param resourceName Location.
207211
* @param executionType ExecutionType.
@@ -214,6 +218,20 @@ public List<? extends Collection<T>> getMatchingRequests(
214218
return client.getMatchingRequests(priority, resourceName,
215219
executionType, capability);
216220
}
221+
222+
/**
223+
* Returns all matching ContainerRequests that match the given
224+
* AllocationRequestId.
225+
*
226+
* NOTE: This matches only requests that were made by the client WITH the
227+
* allocationRequestId specified.
228+
*
229+
* @param allocationRequestId AllocationRequestId.
230+
* @return All matching ContainerRequests
231+
*/
232+
public Collection<T> getMatchingRequests(long allocationRequestId) {
233+
return client.getMatchingRequests(allocationRequestId);
234+
}
217235

218236
/**
219237
* Registers this application master with the resource manager. On successful

0 commit comments

Comments
 (0)