YARN-918. Remove ApplicationAttemptId from RegisterApplicationMasterRequestProto. Contributed by Vinod K V.
git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1504735 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
7a29bccd7a
commit
ac914f79bc
|
@ -93,9 +93,9 @@ public class LocalContainerAllocator extends RMCommunicator
|
|||
@SuppressWarnings("unchecked")
|
||||
@Override
|
||||
protected synchronized void heartbeat() throws Exception {
|
||||
AllocateRequest allocateRequest = AllocateRequest.newInstance(
|
||||
this.applicationAttemptId, this.lastResponseID, super
|
||||
.getApplicationProgress(), new ArrayList<ResourceRequest>(),
|
||||
AllocateRequest allocateRequest =
|
||||
AllocateRequest.newInstance(this.lastResponseID,
|
||||
super.getApplicationProgress(), new ArrayList<ResourceRequest>(),
|
||||
new ArrayList<ContainerId>(), null);
|
||||
AllocateResponse allocateResponse;
|
||||
try {
|
||||
|
@ -143,7 +143,7 @@ public class LocalContainerAllocator extends RMCommunicator
|
|||
LOG.info("Processing the event " + event.toString());
|
||||
// Assign the same container ID as the AM
|
||||
ContainerId cID =
|
||||
ContainerId.newInstance(applicationAttemptId,
|
||||
ContainerId.newInstance(getContext().getApplicationAttemptId(),
|
||||
this.containerId.getId());
|
||||
Container container = recordFactory.newRecordInstance(Container.class);
|
||||
container.setId(cID);
|
||||
|
|
|
@ -45,7 +45,6 @@ import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest
|
|||
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRequest;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
||||
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
|
@ -63,7 +62,6 @@ public abstract class RMCommunicator extends AbstractService
|
|||
private static final Log LOG = LogFactory.getLog(RMContainerAllocator.class);
|
||||
private int rmPollInterval;//millis
|
||||
protected ApplicationId applicationId;
|
||||
protected ApplicationAttemptId applicationAttemptId;
|
||||
private final AtomicBoolean stopped;
|
||||
protected Thread allocatorThread;
|
||||
@SuppressWarnings("rawtypes")
|
||||
|
@ -91,7 +89,6 @@ public abstract class RMCommunicator extends AbstractService
|
|||
this.context = context;
|
||||
this.eventHandler = context.getEventHandler();
|
||||
this.applicationId = context.getApplicationID();
|
||||
this.applicationAttemptId = context.getApplicationAttemptId();
|
||||
this.stopped = new AtomicBoolean(false);
|
||||
this.heartbeatCallbacks = new ConcurrentLinkedQueue<Runnable>();
|
||||
}
|
||||
|
@ -142,7 +139,6 @@ public abstract class RMCommunicator extends AbstractService
|
|||
try {
|
||||
RegisterApplicationMasterRequest request =
|
||||
recordFactory.newRecordInstance(RegisterApplicationMasterRequest.class);
|
||||
request.setApplicationAttemptId(applicationAttemptId);
|
||||
if (serviceAddr != null) {
|
||||
request.setHost(serviceAddr.getHostName());
|
||||
request.setRpcPort(serviceAddr.getPort());
|
||||
|
@ -193,11 +189,8 @@ public abstract class RMCommunicator extends AbstractService
|
|||
LOG.info("History url is " + historyUrl);
|
||||
|
||||
FinishApplicationMasterRequest request =
|
||||
recordFactory.newRecordInstance(FinishApplicationMasterRequest.class);
|
||||
request.setAppAttemptId(this.applicationAttemptId);
|
||||
request.setFinalApplicationStatus(finishState);
|
||||
request.setDiagnostics(sb.toString());
|
||||
request.setTrackingUrl(historyUrl);
|
||||
FinishApplicationMasterRequest.newInstance(finishState,
|
||||
sb.toString(), historyUrl);
|
||||
scheduler.finishApplicationMaster(request);
|
||||
} catch(Exception are) {
|
||||
LOG.error("Exception while unregistering ", are);
|
||||
|
|
|
@ -1161,14 +1161,6 @@ public class RMContainerAllocator extends RMContainerRequestor
|
|||
return containerToAttemptMap.get(cId);
|
||||
}
|
||||
|
||||
NodeId getNodeId(TaskAttemptId tId) {
|
||||
if (tId.getTaskId().getTaskType().equals(TaskType.MAP)) {
|
||||
return maps.get(tId).getNodeId();
|
||||
} else {
|
||||
return reduces.get(tId).getNodeId();
|
||||
}
|
||||
}
|
||||
|
||||
ContainerId get(TaskAttemptId tId) {
|
||||
Container taskContainer;
|
||||
if (tId.getTaskId().getTaskType().equals(TaskType.MAP)) {
|
||||
|
|
|
@ -145,10 +145,10 @@ public abstract class RMContainerRequestor extends RMCommunicator {
|
|||
}
|
||||
|
||||
protected AllocateResponse makeRemoteRequest() throws IOException {
|
||||
AllocateRequest allocateRequest = AllocateRequest.newInstance(
|
||||
applicationAttemptId, lastResponseID, super.getApplicationProgress(),
|
||||
new ArrayList<ResourceRequest>(ask), new ArrayList<ContainerId>(
|
||||
release), null);
|
||||
AllocateRequest allocateRequest =
|
||||
AllocateRequest.newInstance(lastResponseID,
|
||||
super.getApplicationProgress(), new ArrayList<ResourceRequest>(ask),
|
||||
new ArrayList<ContainerId>(release), null);
|
||||
AllocateResponse allocateResponse;
|
||||
try {
|
||||
allocateResponse = scheduler.allocate(allocateRequest);
|
||||
|
|
|
@ -234,7 +234,7 @@ public class MRAppBenchmark {
|
|||
for (int i = 0; i < numContainers; i++) {
|
||||
ContainerId containerId =
|
||||
ContainerId.newInstance(
|
||||
request.getApplicationAttemptId(),
|
||||
getContext().getApplicationAttemptId(),
|
||||
request.getResponseId() + i);
|
||||
containers.add(Container.newInstance(containerId,
|
||||
NodeId.newInstance("host" + containerId.getId(), 2345),
|
||||
|
|
|
@ -256,6 +256,9 @@ Release 2.1.0-beta - 2013-07-02
|
|||
YARN-701. Use application tokens irrespective of secure or non-secure
|
||||
mode. (vinodkv via acmurthy)
|
||||
|
||||
YARN-918. Remove ApplicationAttemptId from
|
||||
RegisterApplicationMasterRequestProto. (vinodkv via acmurthy)
|
||||
|
||||
NEW FEATURES
|
||||
|
||||
YARN-482. FS: Extend SchedulingMode to intermediate queues.
|
||||
|
|
|
@ -23,10 +23,9 @@ import java.util.List;
|
|||
import org.apache.hadoop.classification.InterfaceAudience.Public;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Stable;
|
||||
import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceBlacklistRequest;
|
||||
import org.apache.hadoop.yarn.api.records.Container;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceBlacklistRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceRequest;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
|
||||
|
@ -36,10 +35,6 @@ import org.apache.hadoop.yarn.util.Records;
|
|||
*
|
||||
* <p>The request includes:
|
||||
* <ul>
|
||||
* <li>
|
||||
* {@link ApplicationAttemptId} being managed by the
|
||||
* <code>ApplicationMaster</code>
|
||||
* </li>
|
||||
* <li>A response id to track duplicate responses.</li>
|
||||
* <li>Progress information.</li>
|
||||
* <li>
|
||||
|
@ -61,13 +56,11 @@ public abstract class AllocateRequest {
|
|||
|
||||
@Public
|
||||
@Stable
|
||||
public static AllocateRequest newInstance(
|
||||
ApplicationAttemptId applicationAttemptId, int responseID,
|
||||
float appProgress, List<ResourceRequest> resourceAsk,
|
||||
public static AllocateRequest newInstance(int responseID, float appProgress,
|
||||
List<ResourceRequest> resourceAsk,
|
||||
List<ContainerId> containersToBeReleased,
|
||||
ResourceBlacklistRequest resourceBlacklistRequest) {
|
||||
AllocateRequest allocateRequest = Records.newRecord(AllocateRequest.class);
|
||||
allocateRequest.setApplicationAttemptId(applicationAttemptId);
|
||||
allocateRequest.setResponseId(responseID);
|
||||
allocateRequest.setProgress(appProgress);
|
||||
allocateRequest.setAskList(resourceAsk);
|
||||
|
@ -76,26 +69,6 @@ public abstract class AllocateRequest {
|
|||
return allocateRequest;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the <code>ApplicationAttemptId</code> being managed by the
|
||||
* <code>ApplicationMaster</code>.
|
||||
* @return <code>ApplicationAttemptId</code> being managed by the
|
||||
* <code>ApplicationMaster</code>
|
||||
*/
|
||||
@Public
|
||||
@Stable
|
||||
public abstract ApplicationAttemptId getApplicationAttemptId();
|
||||
|
||||
/**
|
||||
* Set the <code>ApplicationAttemptId</code> being managed by the
|
||||
* <code>ApplicationMaster</code>.
|
||||
* @param applicationAttemptId <code>ApplicationAttemptId</code> being managed
|
||||
* by the <code>ApplicationMaster</code>
|
||||
*/
|
||||
@Public
|
||||
@Stable
|
||||
public abstract void setApplicationAttemptId(ApplicationAttemptId applicationAttemptId);
|
||||
|
||||
/**
|
||||
* Get the <em>response id</em> used to track duplicate responses.
|
||||
* @return <em>response id</em>
|
||||
|
|
|
@ -21,7 +21,6 @@ package org.apache.hadoop.yarn.api.protocolrecords;
|
|||
import org.apache.hadoop.classification.InterfaceAudience.Public;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Stable;
|
||||
import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
|
||||
|
@ -31,10 +30,6 @@ import org.apache.hadoop.yarn.util.Records;
|
|||
*
|
||||
* <p>The final request includes details such:
|
||||
* <ul>
|
||||
* <li>
|
||||
* {@link ApplicationAttemptId} being managed by the
|
||||
* <code>ApplicationMaster</code>
|
||||
* </li>
|
||||
* <li>Final state of the <code>ApplicationMaster</code></li>
|
||||
* <li>
|
||||
* Diagnostic information in case of failure of the
|
||||
|
@ -53,37 +48,15 @@ public abstract class FinishApplicationMasterRequest {
|
|||
@Public
|
||||
@Stable
|
||||
public static FinishApplicationMasterRequest newInstance(
|
||||
ApplicationAttemptId appAttemptId, FinalApplicationStatus finalAppStatus,
|
||||
String diagnostics, String url) {
|
||||
FinalApplicationStatus finalAppStatus, String diagnostics, String url) {
|
||||
FinishApplicationMasterRequest request =
|
||||
Records.newRecord(FinishApplicationMasterRequest.class);
|
||||
request.setAppAttemptId(appAttemptId);
|
||||
request.setFinalApplicationStatus(finalAppStatus);
|
||||
request.setDiagnostics(diagnostics);
|
||||
request.setTrackingUrl(url);
|
||||
return request;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the <code>ApplicationAttemptId</code> being managed by the
|
||||
* <code>ApplicationMaster</code>.
|
||||
* @return <code>ApplicationAttemptId</code> being managed by the
|
||||
* <code>ApplicationMaster</code>
|
||||
*/
|
||||
@Public
|
||||
@Stable
|
||||
public abstract ApplicationAttemptId getApplicationAttemptId();
|
||||
|
||||
/**
|
||||
* Set the <code>ApplicationAttemptId</code> being managed by the
|
||||
* <code>ApplicationMaster</code>.
|
||||
* @param applicationAttemptId <code>ApplicationAttemptId</code> being managed
|
||||
* by the <code>ApplicationMaster</code>
|
||||
*/
|
||||
@Public
|
||||
@Stable
|
||||
public abstract void setAppAttemptId(ApplicationAttemptId applicationAttemptId);
|
||||
|
||||
/**
|
||||
* Get <em>final state</em> of the <code>ApplicationMaster</code>.
|
||||
* @return <em>final state</em> of the <code>ApplicationMaster</code>
|
||||
|
|
|
@ -21,7 +21,6 @@ package org.apache.hadoop.yarn.api.protocolrecords;
|
|||
import org.apache.hadoop.classification.InterfaceAudience.Public;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Stable;
|
||||
import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
|
||||
/**
|
||||
|
@ -30,10 +29,6 @@ import org.apache.hadoop.yarn.util.Records;
|
|||
*
|
||||
* <p>The registration includes details such as:
|
||||
* <ul>
|
||||
* <li>
|
||||
* {@link ApplicationAttemptId} being managed by the
|
||||
* <code>ApplicationMaster</code>
|
||||
* </li>
|
||||
* <li>Hostname on which the AM is running.</li>
|
||||
* <li>RPC Port</li>
|
||||
* <li>Tracking URL</li>
|
||||
|
@ -57,38 +52,16 @@ public abstract class RegisterApplicationMasterRequest {
|
|||
*/
|
||||
@Public
|
||||
@Stable
|
||||
public static RegisterApplicationMasterRequest newInstance(
|
||||
ApplicationAttemptId applicationAttemptId, String host, int port,
|
||||
String trackingUrl) {
|
||||
public static RegisterApplicationMasterRequest newInstance(String host,
|
||||
int port, String trackingUrl) {
|
||||
RegisterApplicationMasterRequest request =
|
||||
Records.newRecord(RegisterApplicationMasterRequest.class);
|
||||
request.setApplicationAttemptId(applicationAttemptId);
|
||||
request.setHost(host);
|
||||
request.setRpcPort(port);
|
||||
request.setTrackingUrl(trackingUrl);
|
||||
return request;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the <code>ApplicationAttemptId</code> being managed by the
|
||||
* <code>ApplicationMaster</code>.
|
||||
* @return <code>ApplicationAttemptId</code> being managed by the
|
||||
* <code>ApplicationMaster</code>
|
||||
*/
|
||||
@Public
|
||||
@Stable
|
||||
public abstract ApplicationAttemptId getApplicationAttemptId();
|
||||
|
||||
/**
|
||||
* Set the <code>ApplicationAttemptId</code> being managed by the
|
||||
* <code>ApplicationMaster</code>.
|
||||
* @param applicationAttemptId <code>ApplicationAttemptId</code> being managed
|
||||
* by the <code>ApplicationMaster</code>
|
||||
*/
|
||||
@Public
|
||||
@Stable
|
||||
public abstract void setApplicationAttemptId(ApplicationAttemptId applicationAttemptId);
|
||||
|
||||
/**
|
||||
* Get the <em>host</em> on which the <code>ApplicationMaster</code> is
|
||||
* running.
|
||||
|
|
|
@ -34,10 +34,9 @@ import "yarn_protos.proto";
|
|||
/////// AM_RM_Protocol ///////////////////////////////
|
||||
//////////////////////////////////////////////////////
|
||||
message RegisterApplicationMasterRequestProto {
|
||||
optional ApplicationAttemptIdProto application_attempt_id = 1;
|
||||
optional string host = 2;
|
||||
optional int32 rpc_port = 3;
|
||||
optional string tracking_url = 4;
|
||||
optional string host = 1;
|
||||
optional int32 rpc_port = 2;
|
||||
optional string tracking_url = 3;
|
||||
}
|
||||
|
||||
message RegisterApplicationMasterResponseProto {
|
||||
|
@ -47,22 +46,20 @@ message RegisterApplicationMasterResponseProto {
|
|||
}
|
||||
|
||||
message FinishApplicationMasterRequestProto {
|
||||
optional ApplicationAttemptIdProto application_attempt_id = 1;
|
||||
optional string diagnostics = 2;
|
||||
optional string tracking_url = 3;
|
||||
optional FinalApplicationStatusProto final_application_status = 4;
|
||||
optional string diagnostics = 1;
|
||||
optional string tracking_url = 2;
|
||||
optional FinalApplicationStatusProto final_application_status = 3;
|
||||
}
|
||||
|
||||
message FinishApplicationMasterResponseProto {
|
||||
}
|
||||
|
||||
message AllocateRequestProto {
|
||||
optional ApplicationAttemptIdProto application_attempt_id = 1;
|
||||
repeated ResourceRequestProto ask = 2;
|
||||
repeated ContainerIdProto release = 3;
|
||||
optional ResourceBlacklistRequestProto blacklist_request = 4;
|
||||
optional int32 response_id = 5;
|
||||
optional float progress = 6;
|
||||
repeated ResourceRequestProto ask = 1;
|
||||
repeated ContainerIdProto release = 2;
|
||||
optional ResourceBlacklistRequestProto blacklist_request = 3;
|
||||
optional int32 response_id = 4;
|
||||
optional float progress = 5;
|
||||
}
|
||||
|
||||
message NMTokenProto {
|
||||
|
|
|
@ -443,7 +443,7 @@ public class ApplicationMaster {
|
|||
|
||||
AMRMClientAsync.CallbackHandler allocListener = new RMCallbackHandler();
|
||||
resourceManager =
|
||||
AMRMClientAsync.createAMRMClientAsync(appAttemptID, 1000, allocListener);
|
||||
AMRMClientAsync.createAMRMClientAsync(1000, allocListener);
|
||||
resourceManager.init(conf);
|
||||
resourceManager.start();
|
||||
|
||||
|
|
|
@ -29,15 +29,15 @@ import org.apache.hadoop.classification.InterfaceStability;
|
|||
import org.apache.hadoop.service.AbstractService;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
|
||||
import org.apache.hadoop.yarn.api.records.Priority;
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.client.api.impl.AMRMClientImpl;
|
||||
import org.apache.hadoop.yarn.exceptions.YarnException;
|
||||
import com.google.common.collect.ImmutableList;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
import com.google.common.collect.ImmutableList;
|
||||
|
||||
@InterfaceAudience.Public
|
||||
@InterfaceStability.Stable
|
||||
|
@ -49,17 +49,13 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
|
|||
* For usage:
|
||||
* <pre>
|
||||
* {@code
|
||||
* AMRMClient.<T>createAMRMClientContainerRequest(appAttemptId)
|
||||
* AMRMClient.<T>createAMRMClientContainerRequest()
|
||||
* }</pre>
|
||||
* @param appAttemptId the appAttemptId associated with the AMRMClient
|
||||
* @return the newly create AMRMClient instance.
|
||||
*/
|
||||
@Public
|
||||
public static <T extends ContainerRequest> AMRMClient<T> createAMRMClient(
|
||||
ApplicationAttemptId appAttemptId) {
|
||||
Preconditions.checkArgument(appAttemptId != null,
|
||||
"ApplicationAttempId should not be null");
|
||||
AMRMClient<T> client = new AMRMClientImpl<T>(appAttemptId);
|
||||
public static <T extends ContainerRequest> AMRMClient<T> createAMRMClient() {
|
||||
AMRMClient<T> client = new AMRMClientImpl<T>();
|
||||
return client;
|
||||
}
|
||||
|
||||
|
|
|
@ -28,7 +28,6 @@ import org.apache.hadoop.classification.InterfaceAudience.Public;
|
|||
import org.apache.hadoop.classification.InterfaceStability.Stable;
|
||||
import org.apache.hadoop.service.AbstractService;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.Container;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerStatus;
|
||||
|
@ -97,24 +96,18 @@ extends AbstractService {
|
|||
protected final AtomicInteger heartbeatIntervalMs = new AtomicInteger();
|
||||
|
||||
public static <T extends ContainerRequest> AMRMClientAsync<T>
|
||||
createAMRMClientAsync(
|
||||
ApplicationAttemptId id,
|
||||
int intervalMs,
|
||||
CallbackHandler callbackHandler) {
|
||||
return new AMRMClientAsyncImpl<T>(id, intervalMs, callbackHandler);
|
||||
createAMRMClientAsync(int intervalMs, CallbackHandler callbackHandler) {
|
||||
return new AMRMClientAsyncImpl<T>(intervalMs, callbackHandler);
|
||||
}
|
||||
|
||||
public static <T extends ContainerRequest> AMRMClientAsync<T>
|
||||
createAMRMClientAsync(
|
||||
AMRMClient<T> client,
|
||||
int intervalMs,
|
||||
createAMRMClientAsync(AMRMClient<T> client, int intervalMs,
|
||||
CallbackHandler callbackHandler) {
|
||||
return new AMRMClientAsyncImpl<T>(client, intervalMs, callbackHandler);
|
||||
}
|
||||
|
||||
protected AMRMClientAsync(ApplicationAttemptId id, int intervalMs,
|
||||
CallbackHandler callbackHandler) {
|
||||
this(new AMRMClientImpl<T>(id), intervalMs, callbackHandler);
|
||||
protected AMRMClientAsync(int intervalMs, CallbackHandler callbackHandler) {
|
||||
this(new AMRMClientImpl<T>(), intervalMs, callbackHandler);
|
||||
}
|
||||
|
||||
@Private
|
||||
|
|
|
@ -32,7 +32,6 @@ import org.apache.hadoop.conf.Configuration;
|
|||
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
|
||||
import org.apache.hadoop.yarn.api.records.AMCommand;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.Container;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerStatus;
|
||||
|
@ -68,9 +67,8 @@ extends AMRMClientAsync<T> {
|
|||
|
||||
private volatile Exception savedException;
|
||||
|
||||
public AMRMClientAsyncImpl(ApplicationAttemptId id, int intervalMs,
|
||||
CallbackHandler callbackHandler) {
|
||||
this(new AMRMClientImpl<T>(id), intervalMs, callbackHandler);
|
||||
public AMRMClientAsyncImpl(int intervalMs, CallbackHandler callbackHandler) {
|
||||
this(new AMRMClientImpl<T>(), intervalMs, callbackHandler);
|
||||
}
|
||||
|
||||
@Private
|
||||
|
|
|
@ -46,7 +46,6 @@ import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
|
|||
import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRequest;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
|
||||
import org.apache.hadoop.yarn.api.records.NMToken;
|
||||
|
@ -61,8 +60,6 @@ import org.apache.hadoop.yarn.client.api.NMTokenCache;
|
|||
import org.apache.hadoop.yarn.conf.YarnConfiguration;
|
||||
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.util.RackResolver;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
|
@ -77,13 +74,9 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
|
|||
private static final List<String> ANY_LIST =
|
||||
Collections.singletonList(ResourceRequest.ANY);
|
||||
|
||||
private final RecordFactory recordFactory =
|
||||
RecordFactoryProvider.getRecordFactory(null);
|
||||
|
||||
private int lastResponseId = 0;
|
||||
|
||||
protected ApplicationMasterProtocol rmClient;
|
||||
protected final ApplicationAttemptId appAttemptId;
|
||||
protected Resource clusterAvailableResources;
|
||||
protected int clusterNodeCount;
|
||||
|
||||
|
@ -154,9 +147,8 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
|
|||
new org.apache.hadoop.yarn.api.records.ResourceRequest.ResourceRequestComparator());
|
||||
protected final Set<ContainerId> release = new TreeSet<ContainerId>();
|
||||
|
||||
public AMRMClientImpl(ApplicationAttemptId appAttemptId) {
|
||||
public AMRMClientImpl() {
|
||||
super(AMRMClientImpl.class.getName());
|
||||
this.appAttemptId = appAttemptId;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
@ -193,18 +185,11 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
|
|||
Preconditions.checkArgument(appHostPort >= 0,
|
||||
"Port number of the host should not be negative");
|
||||
// do this only once ???
|
||||
RegisterApplicationMasterRequest request = recordFactory
|
||||
.newRecordInstance(RegisterApplicationMasterRequest.class);
|
||||
synchronized (this) {
|
||||
request.setApplicationAttemptId(appAttemptId);
|
||||
}
|
||||
request.setHost(appHostName);
|
||||
request.setRpcPort(appHostPort);
|
||||
if(appTrackingUrl != null) {
|
||||
request.setTrackingUrl(appTrackingUrl);
|
||||
}
|
||||
RegisterApplicationMasterResponse response = rmClient
|
||||
.registerApplicationMaster(request);
|
||||
RegisterApplicationMasterRequest request =
|
||||
RegisterApplicationMasterRequest.newInstance(appHostName, appHostPort,
|
||||
appTrackingUrl);
|
||||
RegisterApplicationMasterResponse response =
|
||||
rmClient.registerApplicationMaster(request);
|
||||
return response;
|
||||
}
|
||||
|
||||
|
@ -233,8 +218,8 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
|
|||
ask.clear();
|
||||
release.clear();
|
||||
allocateRequest =
|
||||
AllocateRequest.newInstance(appAttemptId, lastResponseId,
|
||||
progressIndicator, askList, releaseList, null);
|
||||
AllocateRequest.newInstance(lastResponseId, progressIndicator,
|
||||
askList, releaseList, null);
|
||||
}
|
||||
|
||||
allocateResponse = rmClient.allocate(allocateRequest);
|
||||
|
@ -292,18 +277,9 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
|
|||
public void unregisterApplicationMaster(FinalApplicationStatus appStatus,
|
||||
String appMessage, String appTrackingUrl) throws YarnException,
|
||||
IOException {
|
||||
Preconditions.checkArgument(appStatus != null,
|
||||
"AppStatus should not be null.");
|
||||
FinishApplicationMasterRequest request = recordFactory
|
||||
.newRecordInstance(FinishApplicationMasterRequest.class);
|
||||
request.setAppAttemptId(appAttemptId);
|
||||
request.setFinalApplicationStatus(appStatus);
|
||||
if(appMessage != null) {
|
||||
request.setDiagnostics(appMessage);
|
||||
}
|
||||
if(appTrackingUrl != null) {
|
||||
request.setTrackingUrl(appTrackingUrl);
|
||||
}
|
||||
FinishApplicationMasterRequest request =
|
||||
FinishApplicationMasterRequest.newInstance(appStatus, appMessage,
|
||||
appTrackingUrl);
|
||||
rmClient.finishApplicationMaster(request);
|
||||
}
|
||||
|
||||
|
@ -553,7 +529,7 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
|
|||
|
||||
if (LOG.isDebugEnabled()) {
|
||||
LOG.debug("addResourceRequest:" + " applicationId="
|
||||
+ appAttemptId + " priority=" + priority.getPriority()
|
||||
+ " priority=" + priority.getPriority()
|
||||
+ " resourceName=" + resourceName + " numContainers="
|
||||
+ resourceRequestInfo.remoteRequest.getNumContainers()
|
||||
+ " #asks=" + ask.size());
|
||||
|
@ -587,7 +563,7 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
|
|||
|
||||
if (LOG.isDebugEnabled()) {
|
||||
LOG.debug("BEFORE decResourceRequest:" + " applicationId="
|
||||
+ appAttemptId + " priority=" + priority.getPriority()
|
||||
+ " priority=" + priority.getPriority()
|
||||
+ " resourceName=" + resourceName + " numContainers="
|
||||
+ resourceRequestInfo.remoteRequest.getNumContainers()
|
||||
+ " #asks=" + ask.size());
|
||||
|
@ -620,7 +596,7 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
|
|||
|
||||
if (LOG.isDebugEnabled()) {
|
||||
LOG.info("AFTER decResourceRequest:" + " applicationId="
|
||||
+ appAttemptId + " priority=" + priority.getPriority()
|
||||
+ " priority=" + priority.getPriority()
|
||||
+ " resourceName=" + resourceName + " numContainers="
|
||||
+ resourceRequestInfo.remoteRequest.getNumContainers()
|
||||
+ " #asks=" + ask.size());
|
||||
|
|
|
@ -200,7 +200,7 @@ public class TestAMRMClient {
|
|||
AMRMClient<ContainerRequest> amClient = null;
|
||||
try {
|
||||
// start am rm client
|
||||
amClient = AMRMClient.<ContainerRequest>createAMRMClient(attemptId);
|
||||
amClient = AMRMClient.<ContainerRequest>createAMRMClient();
|
||||
amClient.init(conf);
|
||||
amClient.start();
|
||||
amClient.registerApplicationMaster("Host", 10000, "");
|
||||
|
@ -314,7 +314,7 @@ public class TestAMRMClient {
|
|||
AMRMClientImpl<ContainerRequest> amClient = null;
|
||||
try {
|
||||
// start am rm client
|
||||
amClient = new AMRMClientImpl<ContainerRequest>(attemptId);
|
||||
amClient = new AMRMClientImpl<ContainerRequest>();
|
||||
amClient.init(conf);
|
||||
amClient.start();
|
||||
amClient.registerApplicationMaster("Host", 10000, "");
|
||||
|
@ -361,7 +361,7 @@ public class TestAMRMClient {
|
|||
// start am rm client
|
||||
amClient =
|
||||
(AMRMClientImpl<ContainerRequest>) AMRMClient
|
||||
.<ContainerRequest> createAMRMClient(attemptId);
|
||||
.<ContainerRequest> createAMRMClient();
|
||||
amClient.init(conf);
|
||||
amClient.start();
|
||||
amClient.registerApplicationMaster("Host", 10000, "");
|
||||
|
@ -482,7 +482,7 @@ public class TestAMRMClient {
|
|||
AMRMClient<ContainerRequest> amClient = null;
|
||||
try {
|
||||
// start am rm client
|
||||
amClient = AMRMClient.<ContainerRequest>createAMRMClient(attemptId);
|
||||
amClient = AMRMClient.<ContainerRequest>createAMRMClient();
|
||||
amClient.init(conf);
|
||||
amClient.start();
|
||||
|
||||
|
|
|
@ -18,30 +18,27 @@
|
|||
|
||||
package org.apache.hadoop.yarn.client.api.impl;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
|
||||
import org.apache.hadoop.net.DNSToSwitchMapping;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
||||
import org.apache.hadoop.yarn.api.records.Priority;
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceRequest;
|
||||
import org.apache.hadoop.yarn.client.api.AMRMClient;
|
||||
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
|
||||
import org.apache.hadoop.yarn.client.api.InvalidContainerRequestException;
|
||||
import org.apache.hadoop.yarn.client.api.impl.AMRMClientImpl;
|
||||
import org.junit.Test;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
public class TestAMRMClientContainerRequest {
|
||||
@Test
|
||||
public void testFillInRacks() {
|
||||
AMRMClientImpl<ContainerRequest> client = new AMRMClientImpl<ContainerRequest>(
|
||||
ApplicationAttemptId.newInstance(ApplicationId.newInstance(0l, 0), 0));
|
||||
AMRMClientImpl<ContainerRequest> client =
|
||||
new AMRMClientImpl<ContainerRequest>();
|
||||
|
||||
Configuration conf = new Configuration();
|
||||
conf.setClass(
|
||||
|
@ -63,8 +60,8 @@ public class TestAMRMClientContainerRequest {
|
|||
|
||||
@Test
|
||||
public void testDisableLocalityRelaxation() {
|
||||
AMRMClientImpl<ContainerRequest> client = new AMRMClientImpl<ContainerRequest>(
|
||||
ApplicationAttemptId.newInstance(ApplicationId.newInstance(0l, 0), 0));
|
||||
AMRMClientImpl<ContainerRequest> client =
|
||||
new AMRMClientImpl<ContainerRequest>();
|
||||
Configuration conf = new Configuration();
|
||||
conf.setClass(
|
||||
CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
||||
|
@ -130,8 +127,8 @@ public class TestAMRMClientContainerRequest {
|
|||
|
||||
@Test (expected = InvalidContainerRequestException.class)
|
||||
public void testDifferentLocalityRelaxationSamePriority() {
|
||||
AMRMClientImpl<ContainerRequest> client = new AMRMClientImpl<ContainerRequest>(
|
||||
ApplicationAttemptId.newInstance(ApplicationId.newInstance(0l, 0), 0));
|
||||
AMRMClientImpl<ContainerRequest> client =
|
||||
new AMRMClientImpl<ContainerRequest>();
|
||||
Configuration conf = new Configuration();
|
||||
conf.setClass(
|
||||
CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
||||
|
@ -151,8 +148,8 @@ public class TestAMRMClientContainerRequest {
|
|||
|
||||
@Test
|
||||
public void testInvalidValidWhenOldRemoved() {
|
||||
AMRMClientImpl<ContainerRequest> client = new AMRMClientImpl<ContainerRequest>(
|
||||
ApplicationAttemptId.newInstance(ApplicationId.newInstance(0l, 0), 0));
|
||||
AMRMClientImpl<ContainerRequest> client =
|
||||
new AMRMClientImpl<ContainerRequest>();
|
||||
Configuration conf = new Configuration();
|
||||
conf.setClass(
|
||||
CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
||||
|
@ -190,8 +187,8 @@ public class TestAMRMClientContainerRequest {
|
|||
|
||||
@Test (expected = InvalidContainerRequestException.class)
|
||||
public void testLocalityRelaxationDifferentLevels() {
|
||||
AMRMClientImpl<ContainerRequest> client = new AMRMClientImpl<ContainerRequest>(
|
||||
ApplicationAttemptId.newInstance(ApplicationId.newInstance(0l, 0), 0));
|
||||
AMRMClientImpl<ContainerRequest> client =
|
||||
new AMRMClientImpl<ContainerRequest>();
|
||||
Configuration conf = new Configuration();
|
||||
conf.setClass(
|
||||
CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
|
||||
|
|
|
@ -157,7 +157,7 @@ public class TestNMClient {
|
|||
// start am rm client
|
||||
rmClient =
|
||||
(AMRMClientImpl<ContainerRequest>) AMRMClient
|
||||
.<ContainerRequest> createAMRMClient(attemptId);
|
||||
.<ContainerRequest> createAMRMClient();
|
||||
rmClient.init(conf);
|
||||
rmClient.start();
|
||||
assertNotNull(rmClient);
|
||||
|
|
|
@ -26,15 +26,12 @@ import java.util.List;
|
|||
import org.apache.hadoop.classification.InterfaceAudience.Private;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceBlacklistRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceRequest;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationAttemptIdPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerIdPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ResourceBlacklistRequestPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ResourceRequestPBImpl;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationAttemptIdProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerIdProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ResourceBlacklistRequestProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ResourceRequestProto;
|
||||
|
@ -48,7 +45,6 @@ public class AllocateRequestPBImpl extends AllocateRequest {
|
|||
AllocateRequestProto.Builder builder = null;
|
||||
boolean viaProto = false;
|
||||
|
||||
private ApplicationAttemptId applicationAttemptID = null;
|
||||
private List<ResourceRequest> ask = null;
|
||||
private List<ContainerId> release = null;
|
||||
private ResourceBlacklistRequest blacklistRequest = null;
|
||||
|
@ -91,9 +87,6 @@ public class AllocateRequestPBImpl extends AllocateRequest {
|
|||
}
|
||||
|
||||
private void mergeLocalToBuilder() {
|
||||
if (this.applicationAttemptID != null) {
|
||||
builder.setApplicationAttemptId(convertToProtoFormat(this.applicationAttemptID));
|
||||
}
|
||||
if (this.ask != null) {
|
||||
addAsksToProto();
|
||||
}
|
||||
|
@ -120,27 +113,6 @@ public class AllocateRequestPBImpl extends AllocateRequest {
|
|||
viaProto = false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ApplicationAttemptId getApplicationAttemptId() {
|
||||
AllocateRequestProtoOrBuilder p = viaProto ? proto : builder;
|
||||
if (this.applicationAttemptID != null) {
|
||||
return this.applicationAttemptID;
|
||||
}
|
||||
if (!p.hasApplicationAttemptId()) {
|
||||
return null;
|
||||
}
|
||||
this.applicationAttemptID = convertFromProtoFormat(p.getApplicationAttemptId());
|
||||
return this.applicationAttemptID;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setApplicationAttemptId(ApplicationAttemptId appAttemptId) {
|
||||
maybeInitBuilder();
|
||||
if (appAttemptId == null)
|
||||
builder.clearApplicationAttemptId();
|
||||
this.applicationAttemptID = appAttemptId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getResponseId() {
|
||||
AllocateRequestProtoOrBuilder p = viaProto ? proto : builder;
|
||||
|
@ -311,14 +283,6 @@ public class AllocateRequestPBImpl extends AllocateRequest {
|
|||
builder.addAllRelease(iterable);
|
||||
}
|
||||
|
||||
private ApplicationAttemptIdPBImpl convertFromProtoFormat(ApplicationAttemptIdProto p) {
|
||||
return new ApplicationAttemptIdPBImpl(p);
|
||||
}
|
||||
|
||||
private ApplicationAttemptIdProto convertToProtoFormat(ApplicationAttemptId t) {
|
||||
return ((ApplicationAttemptIdPBImpl)t).getProto();
|
||||
}
|
||||
|
||||
private ResourceRequestPBImpl convertFromProtoFormat(ResourceRequestProto p) {
|
||||
return new ResourceRequestPBImpl(p);
|
||||
}
|
||||
|
|
|
@ -22,11 +22,8 @@ package org.apache.hadoop.yarn.api.protocolrecords.impl.pb;
|
|||
import org.apache.hadoop.classification.InterfaceAudience.Private;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationAttemptIdPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ProtoUtils;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationAttemptIdProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.FinalApplicationStatusProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnServiceProtos.FinishApplicationMasterRequestProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnServiceProtos.FinishApplicationMasterRequestProtoOrBuilder;
|
||||
|
@ -38,9 +35,6 @@ public class FinishApplicationMasterRequestPBImpl extends FinishApplicationMaste
|
|||
FinishApplicationMasterRequestProto.Builder builder = null;
|
||||
boolean viaProto = false;
|
||||
|
||||
private ApplicationAttemptId appAttemptId = null;
|
||||
|
||||
|
||||
public FinishApplicationMasterRequestPBImpl() {
|
||||
builder = FinishApplicationMasterRequestProto.newBuilder();
|
||||
}
|
||||
|
@ -78,9 +72,6 @@ public class FinishApplicationMasterRequestPBImpl extends FinishApplicationMaste
|
|||
}
|
||||
|
||||
private void mergeLocalToBuilder() {
|
||||
if (this.appAttemptId != null) {
|
||||
builder.setApplicationAttemptId(convertToProtoFormat(this.appAttemptId));
|
||||
}
|
||||
}
|
||||
|
||||
private void mergeLocalToProto() {
|
||||
|
@ -98,27 +89,6 @@ public class FinishApplicationMasterRequestPBImpl extends FinishApplicationMaste
|
|||
viaProto = false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ApplicationAttemptId getApplicationAttemptId() {
|
||||
FinishApplicationMasterRequestProtoOrBuilder p = viaProto ? proto : builder;
|
||||
if (this.appAttemptId != null) {
|
||||
return this.appAttemptId;
|
||||
}
|
||||
if (!p.hasApplicationAttemptId()) {
|
||||
return null;
|
||||
}
|
||||
this.appAttemptId = convertFromProtoFormat(p.getApplicationAttemptId());
|
||||
return this.appAttemptId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setAppAttemptId(ApplicationAttemptId applicationAttemptId) {
|
||||
maybeInitBuilder();
|
||||
if (applicationAttemptId == null)
|
||||
builder.clearApplicationAttemptId();
|
||||
this.appAttemptId = applicationAttemptId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getDiagnostics() {
|
||||
FinishApplicationMasterRequestProtoOrBuilder p = viaProto ? proto : builder;
|
||||
|
@ -128,6 +98,10 @@ public class FinishApplicationMasterRequestPBImpl extends FinishApplicationMaste
|
|||
@Override
|
||||
public void setDiagnostics(String diagnostics) {
|
||||
maybeInitBuilder();
|
||||
if (diagnostics == null) {
|
||||
builder.clearDiagnostics();
|
||||
return;
|
||||
}
|
||||
builder.setDiagnostics(diagnostics);
|
||||
}
|
||||
|
||||
|
@ -140,6 +114,10 @@ public class FinishApplicationMasterRequestPBImpl extends FinishApplicationMaste
|
|||
@Override
|
||||
public void setTrackingUrl(String url) {
|
||||
maybeInitBuilder();
|
||||
if (url == null) {
|
||||
builder.clearTrackingUrl();
|
||||
return;
|
||||
}
|
||||
builder.setTrackingUrl(url);
|
||||
}
|
||||
|
||||
|
@ -162,14 +140,6 @@ public class FinishApplicationMasterRequestPBImpl extends FinishApplicationMaste
|
|||
builder.setFinalApplicationStatus(convertToProtoFormat(finalState));
|
||||
}
|
||||
|
||||
private ApplicationAttemptIdPBImpl convertFromProtoFormat(ApplicationAttemptIdProto p) {
|
||||
return new ApplicationAttemptIdPBImpl(p);
|
||||
}
|
||||
|
||||
private ApplicationAttemptIdProto convertToProtoFormat(ApplicationAttemptId t) {
|
||||
return ((ApplicationAttemptIdPBImpl)t).getProto();
|
||||
}
|
||||
|
||||
private FinalApplicationStatus convertFromProtoFormat(FinalApplicationStatusProto s) {
|
||||
return ProtoUtils.convertFromProtoFormat(s);
|
||||
}
|
||||
|
|
|
@ -22,9 +22,6 @@ package org.apache.hadoop.yarn.api.protocolrecords.impl.pb;
|
|||
import org.apache.hadoop.classification.InterfaceAudience.Private;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationAttemptIdPBImpl;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationAttemptIdProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnServiceProtos.RegisterApplicationMasterRequestProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnServiceProtos.RegisterApplicationMasterRequestProtoOrBuilder;
|
||||
|
||||
|
@ -35,9 +32,6 @@ public class RegisterApplicationMasterRequestPBImpl extends RegisterApplicationM
|
|||
RegisterApplicationMasterRequestProto.Builder builder = null;
|
||||
boolean viaProto = false;
|
||||
|
||||
private ApplicationAttemptId applicationAttemptId = null;
|
||||
|
||||
|
||||
public RegisterApplicationMasterRequestPBImpl() {
|
||||
builder = RegisterApplicationMasterRequestProto.newBuilder();
|
||||
}
|
||||
|
@ -75,9 +69,6 @@ public class RegisterApplicationMasterRequestPBImpl extends RegisterApplicationM
|
|||
}
|
||||
|
||||
private void mergeLocalToBuilder() {
|
||||
if (this.applicationAttemptId != null && !((ApplicationAttemptIdPBImpl)this.applicationAttemptId).getProto().equals(builder.getApplicationAttemptId())) {
|
||||
builder.setApplicationAttemptId(convertToProtoFormat(this.applicationAttemptId));
|
||||
}
|
||||
}
|
||||
|
||||
private void mergeLocalToProto() {
|
||||
|
@ -95,28 +86,6 @@ public class RegisterApplicationMasterRequestPBImpl extends RegisterApplicationM
|
|||
viaProto = false;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public ApplicationAttemptId getApplicationAttemptId() {
|
||||
RegisterApplicationMasterRequestProtoOrBuilder p = viaProto ? proto : builder;
|
||||
if (this.applicationAttemptId != null) {
|
||||
return this.applicationAttemptId;
|
||||
}
|
||||
if (!p.hasApplicationAttemptId()) {
|
||||
return null;
|
||||
}
|
||||
this.applicationAttemptId = convertFromProtoFormat(p.getApplicationAttemptId());
|
||||
return this.applicationAttemptId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setApplicationAttemptId(ApplicationAttemptId applicationMaster) {
|
||||
maybeInitBuilder();
|
||||
if (applicationMaster == null)
|
||||
builder.clearApplicationAttemptId();
|
||||
this.applicationAttemptId = applicationMaster;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getHost() {
|
||||
RegisterApplicationMasterRequestProtoOrBuilder p = viaProto ? proto : builder;
|
||||
|
@ -126,6 +95,10 @@ public class RegisterApplicationMasterRequestPBImpl extends RegisterApplicationM
|
|||
@Override
|
||||
public void setHost(String host) {
|
||||
maybeInitBuilder();
|
||||
if (host == null) {
|
||||
builder.clearHost();
|
||||
return;
|
||||
}
|
||||
builder.setHost(host);
|
||||
}
|
||||
|
||||
|
@ -150,15 +123,10 @@ public class RegisterApplicationMasterRequestPBImpl extends RegisterApplicationM
|
|||
@Override
|
||||
public void setTrackingUrl(String url) {
|
||||
maybeInitBuilder();
|
||||
if (url == null) {
|
||||
builder.clearTrackingUrl();
|
||||
return;
|
||||
}
|
||||
builder.setTrackingUrl(url);
|
||||
}
|
||||
|
||||
private ApplicationAttemptIdPBImpl convertFromProtoFormat(ApplicationAttemptIdProto p) {
|
||||
return new ApplicationAttemptIdPBImpl(p);
|
||||
}
|
||||
|
||||
private ApplicationAttemptIdProto convertToProtoFormat(ApplicationAttemptId t) {
|
||||
return ((ApplicationAttemptIdPBImpl)t).getProto();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
@ -31,7 +31,6 @@ import org.apache.hadoop.classification.InterfaceAudience.Private;
|
|||
import org.apache.hadoop.io.Text;
|
||||
import org.apache.hadoop.net.NetUtils;
|
||||
import org.apache.hadoop.security.SecurityUtil;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
|
||||
import org.apache.hadoop.yarn.api.records.AMCommand;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
|
||||
|
@ -391,20 +390,6 @@ public class BuilderUtils {
|
|||
return url;
|
||||
}
|
||||
|
||||
public static AllocateRequest newAllocateRequest(
|
||||
ApplicationAttemptId applicationAttemptId, int responseID,
|
||||
float appProgress, List<ResourceRequest> resourceAsk,
|
||||
List<ContainerId> containersToBeReleased) {
|
||||
AllocateRequest allocateRequest = recordFactory
|
||||
.newRecordInstance(AllocateRequest.class);
|
||||
allocateRequest.setApplicationAttemptId(applicationAttemptId);
|
||||
allocateRequest.setResponseId(responseID);
|
||||
allocateRequest.setProgress(appProgress);
|
||||
allocateRequest.setAskList(resourceAsk);
|
||||
allocateRequest.setReleaseList(containersToBeReleased);
|
||||
return allocateRequest;
|
||||
}
|
||||
|
||||
public static AllocateResponse newAllocateResponse(int responseId,
|
||||
List<ContainerStatus> completedContainers,
|
||||
List<Container> allocatedContainers, List<NodeReport> updatedNodes,
|
||||
|
|
|
@ -167,18 +167,16 @@ public class ApplicationMasterService extends AbstractService implements
|
|||
return result;
|
||||
}
|
||||
|
||||
private void authorizeRequest(ApplicationAttemptId appAttemptID)
|
||||
private ApplicationAttemptId authorizeRequest()
|
||||
throws YarnException {
|
||||
|
||||
String appAttemptIDStr = appAttemptID.toString();
|
||||
|
||||
UserGroupInformation remoteUgi;
|
||||
try {
|
||||
remoteUgi = UserGroupInformation.getCurrentUser();
|
||||
} catch (IOException e) {
|
||||
String msg = "Cannot obtain the user-name for ApplicationAttemptID: "
|
||||
+ appAttemptIDStr + ". Got exception: "
|
||||
+ StringUtils.stringifyException(e);
|
||||
String msg =
|
||||
"Cannot obtain the user-name for authorizing ApplicationMaster. "
|
||||
+ "Got exception: " + StringUtils.stringifyException(e);
|
||||
LOG.warn(msg);
|
||||
throw RPCUtil.getRemoteException(msg);
|
||||
}
|
||||
|
@ -190,14 +188,15 @@ public class ApplicationMasterService extends AbstractService implements
|
|||
appTokenIdentifier = selectAMRMTokenIdentifier(remoteUgi);
|
||||
if (appTokenIdentifier == null) {
|
||||
tokenFound = false;
|
||||
message = "No AMRMToken found for " + appAttemptIDStr;
|
||||
message = "No AMRMToken found for user " + remoteUgi.getUserName();
|
||||
} else {
|
||||
tokenFound = true;
|
||||
}
|
||||
} catch (IOException e) {
|
||||
tokenFound = false;
|
||||
message =
|
||||
"Got exception while looking for AMRMToken for " + appAttemptIDStr;
|
||||
"Got exception while looking for AMRMToken for user "
|
||||
+ remoteUgi.getUserName();
|
||||
}
|
||||
|
||||
if (!tokenFound) {
|
||||
|
@ -205,15 +204,7 @@ public class ApplicationMasterService extends AbstractService implements
|
|||
throw RPCUtil.getRemoteException(message);
|
||||
}
|
||||
|
||||
ApplicationAttemptId remoteApplicationAttemptId =
|
||||
appTokenIdentifier.getApplicationAttemptId();
|
||||
if (!remoteApplicationAttemptId.equals(appAttemptID)) {
|
||||
String msg = "Unauthorized request from ApplicationMaster. "
|
||||
+ "Expected ApplicationAttemptID: " + remoteApplicationAttemptId
|
||||
+ " Found: " + appAttemptIDStr;
|
||||
LOG.warn(msg);
|
||||
throw RPCUtil.getRemoteException(msg);
|
||||
}
|
||||
return appTokenIdentifier.getApplicationAttemptId();
|
||||
}
|
||||
|
||||
@Override
|
||||
|
@ -221,9 +212,7 @@ public class ApplicationMasterService extends AbstractService implements
|
|||
RegisterApplicationMasterRequest request) throws YarnException,
|
||||
IOException {
|
||||
|
||||
ApplicationAttemptId applicationAttemptId = request
|
||||
.getApplicationAttemptId();
|
||||
authorizeRequest(applicationAttemptId);
|
||||
ApplicationAttemptId applicationAttemptId = authorizeRequest();
|
||||
|
||||
ApplicationId appID = applicationAttemptId.getApplicationId();
|
||||
AllocateResponse lastResponse = responseMap.get(applicationAttemptId);
|
||||
|
@ -293,9 +282,7 @@ public class ApplicationMasterService extends AbstractService implements
|
|||
FinishApplicationMasterRequest request) throws YarnException,
|
||||
IOException {
|
||||
|
||||
ApplicationAttemptId applicationAttemptId = request
|
||||
.getApplicationAttemptId();
|
||||
authorizeRequest(applicationAttemptId);
|
||||
ApplicationAttemptId applicationAttemptId = authorizeRequest();
|
||||
|
||||
AllocateResponse lastResponse = responseMap.get(applicationAttemptId);
|
||||
if (lastResponse == null) {
|
||||
|
@ -343,8 +330,7 @@ public class ApplicationMasterService extends AbstractService implements
|
|||
public AllocateResponse allocate(AllocateRequest request)
|
||||
throws YarnException, IOException {
|
||||
|
||||
ApplicationAttemptId appAttemptId = request.getApplicationAttemptId();
|
||||
authorizeRequest(appAttemptId);
|
||||
ApplicationAttemptId appAttemptId = authorizeRequest();
|
||||
|
||||
this.amLivelinessMonitor.receivedPing(appAttemptId);
|
||||
|
||||
|
|
|
@ -96,7 +96,6 @@ public class MockAM {
|
|||
responseId = 0;
|
||||
final RegisterApplicationMasterRequest req =
|
||||
Records.newRecord(RegisterApplicationMasterRequest.class);
|
||||
req.setApplicationAttemptId(attemptId);
|
||||
req.setHost("");
|
||||
req.setRpcPort(1);
|
||||
req.setTrackingUrl("");
|
||||
|
@ -174,8 +173,9 @@ public class MockAM {
|
|||
public AllocateResponse allocate(
|
||||
List<ResourceRequest> resourceRequest, List<ContainerId> releases)
|
||||
throws Exception {
|
||||
final AllocateRequest req = AllocateRequest.newInstance(attemptId,
|
||||
++responseId, 0F, resourceRequest, releases, null);
|
||||
final AllocateRequest req =
|
||||
AllocateRequest.newInstance(++responseId, 0F, resourceRequest,
|
||||
releases, null);
|
||||
UserGroupInformation ugi =
|
||||
UserGroupInformation.createRemoteUser(attemptId.toString());
|
||||
Token<AMRMTokenIdentifier> token =
|
||||
|
@ -197,11 +197,8 @@ public class MockAM {
|
|||
public void unregisterAppAttempt() throws Exception {
|
||||
waitForState(RMAppAttemptState.RUNNING);
|
||||
final FinishApplicationMasterRequest req =
|
||||
Records.newRecord(FinishApplicationMasterRequest.class);
|
||||
req.setAppAttemptId(attemptId);
|
||||
req.setDiagnostics("");
|
||||
req.setFinalApplicationStatus(FinalApplicationStatus.SUCCEEDED);
|
||||
req.setTrackingUrl("");
|
||||
FinishApplicationMasterRequest.newInstance(
|
||||
FinalApplicationStatus.SUCCEEDED, "", "");
|
||||
UserGroupInformation ugi =
|
||||
UserGroupInformation.createRemoteUser(attemptId.toString());
|
||||
Token<AMRMTokenIdentifier> token =
|
||||
|
|
|
@ -52,8 +52,8 @@ import org.apache.hadoop.yarn.ipc.YarnRPC;
|
|||
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
|
||||
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.utils.BuilderUtils;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
import org.junit.After;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
|
@ -66,6 +66,7 @@ public class TestAMAuthorization {
|
|||
private static final Log LOG = LogFactory.getLog(TestAMAuthorization.class);
|
||||
|
||||
private final Configuration conf;
|
||||
private MockRM rm;
|
||||
|
||||
@Parameters
|
||||
public static Collection<Object[]> configs() {
|
||||
|
@ -82,6 +83,13 @@ public class TestAMAuthorization {
|
|||
UserGroupInformation.setConfiguration(conf);
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown() {
|
||||
if (rm != null) {
|
||||
rm.stop();
|
||||
}
|
||||
}
|
||||
|
||||
public static final class MyContainerManager implements ContainerManagementProtocol {
|
||||
|
||||
public ByteBuffer containerTokens;
|
||||
|
@ -139,7 +147,7 @@ public class TestAMAuthorization {
|
|||
@Test
|
||||
public void testAuthorizedAccess() throws Exception {
|
||||
MyContainerManager containerManager = new MyContainerManager();
|
||||
final MockRM rm =
|
||||
rm =
|
||||
new MockRMWithAMS(conf, containerManager);
|
||||
rm.start();
|
||||
|
||||
|
@ -183,7 +191,6 @@ public class TestAMAuthorization {
|
|||
|
||||
RegisterApplicationMasterRequest request = Records
|
||||
.newRecord(RegisterApplicationMasterRequest.class);
|
||||
request.setApplicationAttemptId(applicationAttemptId);
|
||||
RegisterApplicationMasterResponse response =
|
||||
client.registerApplicationMaster(request);
|
||||
Assert.assertNotNull(response.getClientToAMTokenMasterKey());
|
||||
|
@ -193,14 +200,12 @@ public class TestAMAuthorization {
|
|||
}
|
||||
Assert.assertEquals("Register response has bad ACLs", "*",
|
||||
response.getApplicationACLs().get(ApplicationAccessType.VIEW_APP));
|
||||
|
||||
rm.stop();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testUnauthorizedAccess() throws Exception {
|
||||
MyContainerManager containerManager = new MyContainerManager();
|
||||
MockRM rm = new MockRMWithAMS(conf, containerManager);
|
||||
rm = new MockRMWithAMS(conf, containerManager);
|
||||
rm.start();
|
||||
|
||||
MockNM nm1 = rm.registerNode("localhost:1234", 5120);
|
||||
|
@ -242,7 +247,6 @@ public class TestAMAuthorization {
|
|||
|
||||
RegisterApplicationMasterRequest request = Records
|
||||
.newRecord(RegisterApplicationMasterRequest.class);
|
||||
request.setApplicationAttemptId(applicationAttemptId);
|
||||
try {
|
||||
client.registerApplicationMaster(request);
|
||||
Assert.fail("Should fail with authorization error");
|
||||
|
@ -260,37 +264,8 @@ public class TestAMAuthorization {
|
|||
+ "Available:" + availableAuthMethods));
|
||||
}
|
||||
|
||||
// Now try to validate invalid authorization.
|
||||
Credentials credentials = containerManager.getContainerCredentials();
|
||||
currentUser.addCredentials(credentials);
|
||||
|
||||
// Create a client to the RM.
|
||||
client = currentUser
|
||||
.doAs(new PrivilegedAction<ApplicationMasterProtocol>() {
|
||||
@Override
|
||||
public ApplicationMasterProtocol run() {
|
||||
return (ApplicationMasterProtocol) rpc.getProxy(ApplicationMasterProtocol.class,
|
||||
serviceAddr, conf);
|
||||
}
|
||||
});
|
||||
|
||||
request =
|
||||
Records.newRecord(RegisterApplicationMasterRequest.class);
|
||||
ApplicationAttemptId otherAppAttemptId = BuilderUtils
|
||||
.newApplicationAttemptId(applicationAttemptId.getApplicationId(), 42);
|
||||
request.setApplicationAttemptId(otherAppAttemptId);
|
||||
try {
|
||||
client.registerApplicationMaster(request);
|
||||
Assert.fail("Should fail with authorization error");
|
||||
} catch (YarnException e) {
|
||||
Assert.assertTrue(e.getMessage().contains(
|
||||
"Unauthorized request from ApplicationMaster. "
|
||||
+ "Expected ApplicationAttemptID: "
|
||||
+ applicationAttemptId.toString() + " Found: "
|
||||
+ otherAppAttemptId.toString()));
|
||||
} finally {
|
||||
rm.stop();
|
||||
}
|
||||
// TODO: Add validation of invalid authorization when there's more data in
|
||||
// the AMRMToken
|
||||
}
|
||||
|
||||
private void waitForLaunchedState(RMAppAttempt attempt)
|
||||
|
|
|
@ -92,8 +92,8 @@ public class TestAMRMRPCNodeUpdates {
|
|||
dispatcher.await();
|
||||
}
|
||||
|
||||
private AllocateResponse allocate(final AllocateRequest req) throws Exception {
|
||||
ApplicationAttemptId attemptId = req.getApplicationAttemptId();
|
||||
private AllocateResponse allocate(final ApplicationAttemptId attemptId,
|
||||
final AllocateRequest req) throws Exception {
|
||||
UserGroupInformation ugi =
|
||||
UserGroupInformation.createRemoteUser(attemptId.toString());
|
||||
Token<AMRMTokenIdentifier> token =
|
||||
|
@ -128,18 +128,20 @@ public class TestAMRMRPCNodeUpdates {
|
|||
am1.registerAppAttempt();
|
||||
|
||||
// allocate request returns no updated node
|
||||
AllocateRequest allocateRequest1 = AllocateRequest.newInstance(attempt1
|
||||
.getAppAttemptId(), 0, 0F, null, null, null);
|
||||
AllocateResponse response1 = allocate(allocateRequest1);
|
||||
AllocateRequest allocateRequest1 =
|
||||
AllocateRequest.newInstance(0, 0F, null, null, null);
|
||||
AllocateResponse response1 =
|
||||
allocate(attempt1.getAppAttemptId(), allocateRequest1);
|
||||
List<NodeReport> updatedNodes = response1.getUpdatedNodes();
|
||||
Assert.assertEquals(0, updatedNodes.size());
|
||||
|
||||
syncNodeHeartbeat(nm4, false);
|
||||
|
||||
// allocate request returns updated node
|
||||
allocateRequest1 = AllocateRequest.newInstance(attempt1
|
||||
.getAppAttemptId(), response1.getResponseId(), 0F, null, null, null);
|
||||
response1 = allocate(allocateRequest1);
|
||||
allocateRequest1 =
|
||||
AllocateRequest.newInstance(response1.getResponseId(), 0F, null, null,
|
||||
null);
|
||||
response1 = allocate(attempt1.getAppAttemptId(), allocateRequest1);
|
||||
updatedNodes = response1.getUpdatedNodes();
|
||||
Assert.assertEquals(1, updatedNodes.size());
|
||||
NodeReport nr = updatedNodes.iterator().next();
|
||||
|
@ -147,7 +149,7 @@ public class TestAMRMRPCNodeUpdates {
|
|||
Assert.assertEquals(NodeState.UNHEALTHY, nr.getNodeState());
|
||||
|
||||
// resending the allocate request returns the same result
|
||||
response1 = allocate(allocateRequest1);
|
||||
response1 = allocate(attempt1.getAppAttemptId(), allocateRequest1);
|
||||
updatedNodes = response1.getUpdatedNodes();
|
||||
Assert.assertEquals(1, updatedNodes.size());
|
||||
nr = updatedNodes.iterator().next();
|
||||
|
@ -157,9 +159,10 @@ public class TestAMRMRPCNodeUpdates {
|
|||
syncNodeLost(nm3);
|
||||
|
||||
// subsequent allocate request returns delta
|
||||
allocateRequest1 = AllocateRequest.newInstance(attempt1
|
||||
.getAppAttemptId(), response1.getResponseId(), 0F, null, null, null);
|
||||
response1 = allocate(allocateRequest1);
|
||||
allocateRequest1 =
|
||||
AllocateRequest.newInstance(response1.getResponseId(), 0F, null, null,
|
||||
null);
|
||||
response1 = allocate(attempt1.getAppAttemptId(), allocateRequest1);
|
||||
updatedNodes = response1.getUpdatedNodes();
|
||||
Assert.assertEquals(1, updatedNodes.size());
|
||||
nr = updatedNodes.iterator().next();
|
||||
|
@ -177,27 +180,30 @@ public class TestAMRMRPCNodeUpdates {
|
|||
am2.registerAppAttempt();
|
||||
|
||||
// allocate request returns no updated node
|
||||
AllocateRequest allocateRequest2 = AllocateRequest.newInstance(attempt2
|
||||
.getAppAttemptId(), 0, 0F, null, null, null);
|
||||
AllocateResponse response2 = allocate(allocateRequest2);
|
||||
AllocateRequest allocateRequest2 =
|
||||
AllocateRequest.newInstance(0, 0F, null, null, null);
|
||||
AllocateResponse response2 =
|
||||
allocate(attempt2.getAppAttemptId(), allocateRequest2);
|
||||
updatedNodes = response2.getUpdatedNodes();
|
||||
Assert.assertEquals(0, updatedNodes.size());
|
||||
|
||||
syncNodeHeartbeat(nm4, true);
|
||||
|
||||
// both AM's should get delta updated nodes
|
||||
allocateRequest1 = AllocateRequest.newInstance(attempt1
|
||||
.getAppAttemptId(), response1.getResponseId(), 0F, null, null, null);
|
||||
response1 = allocate(allocateRequest1);
|
||||
allocateRequest1 =
|
||||
AllocateRequest.newInstance(response1.getResponseId(), 0F, null, null,
|
||||
null);
|
||||
response1 = allocate(attempt1.getAppAttemptId(), allocateRequest1);
|
||||
updatedNodes = response1.getUpdatedNodes();
|
||||
Assert.assertEquals(1, updatedNodes.size());
|
||||
nr = updatedNodes.iterator().next();
|
||||
Assert.assertEquals(nm4.getNodeId(), nr.getNodeId());
|
||||
Assert.assertEquals(NodeState.RUNNING, nr.getNodeState());
|
||||
|
||||
allocateRequest2 = AllocateRequest.newInstance(attempt2
|
||||
.getAppAttemptId(), response2.getResponseId(), 0F, null, null, null);
|
||||
response2 = allocate(allocateRequest2);
|
||||
allocateRequest2 =
|
||||
AllocateRequest.newInstance(response2.getResponseId(), 0F, null, null,
|
||||
null);
|
||||
response2 = allocate(attempt2.getAppAttemptId(), allocateRequest2);
|
||||
updatedNodes = response2.getUpdatedNodes();
|
||||
Assert.assertEquals(1, updatedNodes.size());
|
||||
nr = updatedNodes.iterator().next();
|
||||
|
@ -205,9 +211,10 @@ public class TestAMRMRPCNodeUpdates {
|
|||
Assert.assertEquals(NodeState.RUNNING, nr.getNodeState());
|
||||
|
||||
// subsequent allocate calls should return no updated nodes
|
||||
allocateRequest2 = AllocateRequest.newInstance(attempt2
|
||||
.getAppAttemptId(), response2.getResponseId(), 0F, null, null, null);
|
||||
response2 = allocate(allocateRequest2);
|
||||
allocateRequest2 =
|
||||
AllocateRequest.newInstance(response2.getResponseId(), 0F, null, null,
|
||||
null);
|
||||
response2 = allocate(attempt2.getAppAttemptId(), allocateRequest2);
|
||||
updatedNodes = response2.getUpdatedNodes();
|
||||
Assert.assertEquals(0, updatedNodes.size());
|
||||
|
||||
|
|
|
@ -57,8 +57,8 @@ public class TestAMRMRPCResponseId {
|
|||
}
|
||||
}
|
||||
|
||||
private AllocateResponse allocate(final AllocateRequest req) throws Exception {
|
||||
ApplicationAttemptId attemptId = req.getApplicationAttemptId();
|
||||
private AllocateResponse allocate(ApplicationAttemptId attemptId,
|
||||
final AllocateRequest req) throws Exception {
|
||||
UserGroupInformation ugi =
|
||||
UserGroupInformation.createRemoteUser(attemptId.toString());
|
||||
org.apache.hadoop.security.token.Token<AMRMTokenIdentifier> token =
|
||||
|
@ -88,25 +88,26 @@ public class TestAMRMRPCResponseId {
|
|||
|
||||
am.registerAppAttempt();
|
||||
|
||||
AllocateRequest allocateRequest = AllocateRequest.newInstance(attempt
|
||||
.getAppAttemptId(), 0, 0F, null, null, null);
|
||||
AllocateRequest allocateRequest =
|
||||
AllocateRequest.newInstance(0, 0F, null, null, null);
|
||||
|
||||
AllocateResponse response = allocate(allocateRequest);
|
||||
AllocateResponse response =
|
||||
allocate(attempt.getAppAttemptId(), allocateRequest);
|
||||
Assert.assertEquals(1, response.getResponseId());
|
||||
Assert.assertTrue(response.getAMCommand() == null);
|
||||
allocateRequest = AllocateRequest.newInstance(attempt
|
||||
.getAppAttemptId(), response.getResponseId(), 0F, null, null, null);
|
||||
allocateRequest =
|
||||
AllocateRequest.newInstance(response.getResponseId(), 0F, null, null,
|
||||
null);
|
||||
|
||||
response = allocate(allocateRequest);
|
||||
response = allocate(attempt.getAppAttemptId(), allocateRequest);
|
||||
Assert.assertEquals(2, response.getResponseId());
|
||||
/* try resending */
|
||||
response = allocate(allocateRequest);
|
||||
response = allocate(attempt.getAppAttemptId(), allocateRequest);
|
||||
Assert.assertEquals(2, response.getResponseId());
|
||||
|
||||
/** try sending old request again **/
|
||||
allocateRequest = AllocateRequest.newInstance(attempt
|
||||
.getAppAttemptId(), 0, 0F, null, null, null);
|
||||
response = allocate(allocateRequest);
|
||||
allocateRequest = AllocateRequest.newInstance(0, 0F, null, null, null);
|
||||
response = allocate(attempt.getAppAttemptId(), allocateRequest);
|
||||
Assert.assertTrue(response.getAMCommand() == AMCommand.AM_RESYNC);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -310,7 +310,6 @@ public class TestSchedulerUtils {
|
|||
|
||||
RegisterApplicationMasterRequest request = Records
|
||||
.newRecord(RegisterApplicationMasterRequest.class);
|
||||
request.setApplicationAttemptId(applicationAttemptId);
|
||||
client.registerApplicationMaster(request);
|
||||
|
||||
ResourceBlacklistRequest blacklistRequest =
|
||||
|
@ -318,8 +317,7 @@ public class TestSchedulerUtils {
|
|||
Collections.singletonList(ResourceRequest.ANY), null);
|
||||
|
||||
AllocateRequest allocateRequest =
|
||||
AllocateRequest.newInstance(applicationAttemptId, 0, 0.0f, null, null,
|
||||
blacklistRequest);
|
||||
AllocateRequest.newInstance(0, 0.0f, null, null, blacklistRequest);
|
||||
boolean error = false;
|
||||
try {
|
||||
client.allocate(allocateRequest);
|
||||
|
|
|
@ -43,7 +43,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.TestAMAuthorization.MockRMW
|
|||
import org.apache.hadoop.yarn.server.resourcemanager.TestAMAuthorization.MyContainerManager;
|
||||
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
|
||||
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
|
||||
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
@ -118,12 +117,10 @@ public class TestAMRMTokens {
|
|||
|
||||
RegisterApplicationMasterRequest request =
|
||||
Records.newRecord(RegisterApplicationMasterRequest.class);
|
||||
request.setApplicationAttemptId(applicationAttemptId);
|
||||
rmClient.registerApplicationMaster(request);
|
||||
|
||||
FinishApplicationMasterRequest finishAMRequest =
|
||||
Records.newRecord(FinishApplicationMasterRequest.class);
|
||||
finishAMRequest.setAppAttemptId(applicationAttemptId);
|
||||
finishAMRequest
|
||||
.setFinalApplicationStatus(FinalApplicationStatus.SUCCEEDED);
|
||||
finishAMRequest.setDiagnostics("diagnostics");
|
||||
|
@ -134,11 +131,8 @@ public class TestAMRMTokens {
|
|||
// exception.
|
||||
rpc.stopProxy(rmClient, conf); // To avoid using cached client
|
||||
rmClient = createRMClient(rm, conf, rpc, currentUser);
|
||||
request.setApplicationAttemptId(BuilderUtils.newApplicationAttemptId(
|
||||
BuilderUtils.newApplicationId(12345, 78), 987));
|
||||
AllocateRequest allocateRequest =
|
||||
Records.newRecord(AllocateRequest.class);
|
||||
allocateRequest.setApplicationAttemptId(applicationAttemptId);
|
||||
try {
|
||||
rmClient.allocate(allocateRequest);
|
||||
Assert.fail("You got to be kidding me! "
|
||||
|
@ -206,13 +200,11 @@ public class TestAMRMTokens {
|
|||
|
||||
RegisterApplicationMasterRequest request =
|
||||
Records.newRecord(RegisterApplicationMasterRequest.class);
|
||||
request.setApplicationAttemptId(applicationAttemptId);
|
||||
rmClient.registerApplicationMaster(request);
|
||||
|
||||
// One allocate call.
|
||||
AllocateRequest allocateRequest =
|
||||
Records.newRecord(AllocateRequest.class);
|
||||
allocateRequest.setApplicationAttemptId(applicationAttemptId);
|
||||
Assert.assertTrue(
|
||||
rmClient.allocate(allocateRequest).getAMCommand() == null);
|
||||
|
||||
|
@ -229,7 +221,6 @@ public class TestAMRMTokens {
|
|||
rpc.stopProxy(rmClient, conf); // To avoid using cached client
|
||||
rmClient = createRMClient(rm, conf, rpc, currentUser);
|
||||
allocateRequest = Records.newRecord(AllocateRequest.class);
|
||||
allocateRequest.setApplicationAttemptId(applicationAttemptId);
|
||||
Assert.assertTrue(
|
||||
rmClient.allocate(allocateRequest).getAMCommand() == null);
|
||||
} finally {
|
||||
|
|
Loading…
Reference in New Issue