You are viewing a plain text version of this content. The canonical link for it is here.
Posted to yarn-commits@hadoop.apache.org by vi...@apache.org on 2013/08/12 23:26:19 UTC
svn commit: r1513258 [2/10] - in
/hadoop/common/branches/YARN-321/hadoop-yarn-project: ./ hadoop-yarn/
hadoop-yarn/conf/ hadoop-yarn/dev-support/ hadoop-yarn/hadoop-yarn-api/
hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/
hadoop-...
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ApplicationReport.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ApplicationReport.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ApplicationReport.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ApplicationReport.java Mon Aug 12 21:25:49 2013
@@ -58,7 +58,7 @@ public abstract class ApplicationReport
YarnApplicationState state, String diagnostics, String url,
long startTime, long finishTime, FinalApplicationStatus finalStatus,
ApplicationResourceUsageReport appResources, String origTrackingUrl,
- float progress, String applicationType) {
+ float progress, String applicationType, Token amRmToken) {
ApplicationReport report = Records.newRecord(ApplicationReport.class);
report.setApplicationId(applicationId);
report.setCurrentApplicationAttemptId(applicationAttemptId);
@@ -78,6 +78,7 @@ public abstract class ApplicationReport
report.setOriginalTrackingUrl(origTrackingUrl);
report.setProgress(progress);
report.setApplicationType(applicationType);
+ report.setAMRMToken(amRmToken);
return report;
}
@@ -319,4 +320,33 @@ public abstract class ApplicationReport
@Private
@Unstable
public abstract void setApplicationType(String applicationType);
+
+ @Private
+ @Stable
+ public abstract void setAMRMToken(Token amRmToken);
+
+ /**
+ * Get the AMRM token of the application.
+ * <p/>
+ * The AMRM token is required for AM to RM scheduling operations. For
+ * managed Application Masters Yarn takes care of injecting it. For unmanaged
+ * Applications Masters, the token must be obtained via this method and set
+ * in the {@link org.apache.hadoop.security.UserGroupInformation} of the
+ * current user.
+ * <p/>
+ * The AMRM token will be returned only if all the following conditions are
+ * met:
+ * <li>
+ * <ul>the requester is the owner of the ApplicationMaster</ul>
+ * <ul>the application master is an unmanaged ApplicationMaster</ul>
+ * <ul>the application master is in ACCEPTED state</ul>
+ * </li>
+ * Else this method returns NULL.
+ *
+ * @return the AM to RM token if available.
+ */
+ @Public
+ @Stable
+ public abstract Token getAMRMToken();
+
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java Mon Aug 12 21:25:49 2013
@@ -57,8 +57,8 @@ import org.apache.hadoop.yarn.util.Recor
* start/stop containers.</p>
*
* @see ApplicationMasterProtocol#allocate(org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest)
- * @see ContainerManagementProtocol#startContainer(org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest)
- * @see ContainerManagementProtocol#stopContainer(org.apache.hadoop.yarn.api.protocolrecords.StopContainerRequest)
+ * @see ContainerManagementProtocol#startContainers(org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest)
+ * @see ContainerManagementProtocol#stopContainers(org.apache.hadoop.yarn.api.protocolrecords.StopContainersRequest)
*/
@Public
@Stable
@@ -155,7 +155,7 @@ public abstract class Container implemen
* <code>Container</code> includes the <code>ContainerToken</code>.</p>
*
* @see ApplicationMasterProtocol#allocate(org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest)
- * @see ContainerManagementProtocol#startContainer(org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest)
+ * @see ContainerManagementProtocol#startContainers(org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest)
*
* @return <code>ContainerToken</code> for the container
*/
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerLaunchContext.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerLaunchContext.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerLaunchContext.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerLaunchContext.java Mon Aug 12 21:25:49 2013
@@ -49,7 +49,7 @@ import org.apache.hadoop.yarn.util.Recor
* </ul>
* </p>
*
- * @see ContainerManagementProtocol#startContainer(org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest)
+ * @see ContainerManagementProtocol#startContainers(org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest)
*/
@Public
@Stable
@@ -132,7 +132,7 @@ public abstract class ContainerLaunchCon
/**
* <p>
- * Get application-specific binary <em>service data</em>. This is a map keyed
+ * Set application-specific binary <em>service data</em>. This is a map keyed
* by the name of each {@link AuxiliaryService} that is configured on a
* NodeManager and value correspond to the application specific data targeted
* for the keyed {@link AuxiliaryService}. All pre-existing Map entries are
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/LocalResource.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/LocalResource.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/LocalResource.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/LocalResource.java Mon Aug 12 21:25:49 2013
@@ -37,7 +37,7 @@ import org.apache.hadoop.yarn.util.Recor
* @see LocalResourceVisibility
* @see ContainerLaunchContext
* @see ApplicationSubmissionContext
- * @see ContainerManagementProtocol#startContainer(org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest)
+ * @see ContainerManagementProtocol#startContainers(org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest)
*/
@Public
@Stable
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/LocalResourceType.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/LocalResourceType.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/LocalResourceType.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/LocalResourceType.java Mon Aug 12 21:25:49 2013
@@ -43,7 +43,7 @@ import org.apache.hadoop.yarn.api.Contai
* @see LocalResource
* @see ContainerLaunchContext
* @see ApplicationSubmissionContext
- * @see ContainerManagementProtocol#startContainer(org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest)
+ * @see ContainerManagementProtocol#startContainers(org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest)
*/
@Public
@Stable
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/LocalResourceVisibility.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/LocalResourceVisibility.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/LocalResourceVisibility.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/LocalResourceVisibility.java Mon Aug 12 21:25:49 2013
@@ -43,7 +43,7 @@ import org.apache.hadoop.yarn.api.Contai
* @see LocalResource
* @see ContainerLaunchContext
* @see ApplicationSubmissionContext
- * @see ContainerManagementProtocol#startContainer(org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest)
+ * @see ContainerManagementProtocol#startContainers(org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest)
*/
@Public
@Stable
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java Mon Aug 12 21:25:49 2013
@@ -655,17 +655,17 @@ public class YarnConfiguration extends C
public static final long DEFAULT_NM_PROCESS_KILL_WAIT_MS =
2000;
- /** Max time to wait to establish a connection to RM when NM starts
+ /** Max time to wait to establish a connection to RM
*/
- public static final String RESOURCEMANAGER_CONNECT_WAIT_SECS =
- NM_PREFIX + "resourcemanager.connect.wait.secs";
- public static final int DEFAULT_RESOURCEMANAGER_CONNECT_WAIT_SECS =
+ public static final String RESOURCEMANAGER_CONNECT_MAX_WAIT_SECS =
+ RM_PREFIX + "resourcemanager.connect.max.wait.secs";
+ public static final int DEFAULT_RESOURCEMANAGER_CONNECT_MAX_WAIT_SECS =
15*60;
- /** Time interval between each NM attempt to connect to RM
+ /** Time interval between each attempt to connect to RM
*/
public static final String RESOURCEMANAGER_CONNECT_RETRY_INTERVAL_SECS =
- NM_PREFIX + "resourcemanager.connect.retry_interval.secs";
+ RM_PREFIX + "resourcemanager.connect.retry_interval.secs";
public static final long DEFAULT_RESOURCEMANAGER_CONNECT_RETRY_INTERVAL_SECS
= 30;
@@ -711,6 +711,14 @@ public class YarnConfiguration extends C
*/
public static boolean DEFAULT_YARN_MINICLUSTER_FIXED_PORTS = false;
+ /**
+ * Whether users are explicitly trying to control resource monitoring
+ * configuration for the MiniYARNCluster. Disabled by default.
+ */
+ public static final String YARN_MINICLUSTER_CONTROL_RESOURCE_MONITORING =
+ YARN_PREFIX + "minicluster.control-resource-monitoring";
+ public static final boolean
+ DEFAULT_YARN_MINICLUSTER_CONTROL_RESOURCE_MONITORING = false;
/** The log directory for the containers */
public static final String YARN_APP_CONTAINER_LOG_DIR =
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/server/api/ApplicationInitializationContext.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/server/api/ApplicationInitializationContext.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/server/api/ApplicationInitializationContext.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/server/api/ApplicationInitializationContext.java Mon Aug 12 21:25:49 2013
@@ -25,7 +25,7 @@ import org.apache.hadoop.classification.
import org.apache.hadoop.classification.InterfaceStability.Evolving;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.yarn.api.ContainerManagementProtocol;
-import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
@@ -70,7 +70,7 @@ public class ApplicationInitializationCo
/**
* Get the data sent to the NodeManager via
- * {@link ContainerManagementProtocol#startContainer(StartContainerRequest)}
+ * {@link ContainerManagementProtocol#startContainers(StartContainersRequest)}
* as part of {@link ContainerLaunchContext#getServiceData()}
*
* @return the servicesData for this application.
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/server/api/AuxiliaryService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/server/api/AuxiliaryService.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/server/api/AuxiliaryService.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/server/api/AuxiliaryService.java Mon Aug 12 21:25:49 2013
@@ -24,8 +24,8 @@ import org.apache.hadoop.classification.
import org.apache.hadoop.classification.InterfaceStability.Evolving;
import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.yarn.api.ContainerManagementProtocol;
-import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.StartContainerResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersResponse;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
/**
@@ -71,8 +71,8 @@ public abstract class AuxiliaryService e
*
* <p>
* The information is passed along to applications via
- * {@link StartContainerResponse#getAllServicesMetaData()} that is returned by
- * {@link ContainerManagementProtocol#startContainer(StartContainerRequest)}
+ * {@link StartContainersResponse#getAllServicesMetaData()} that is returned by
+ * {@link ContainerManagementProtocol#startContainers(StartContainersRequest)}
* </p>
*
* @return meta-data for this service that should be made available to
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/containermanagement_protocol.proto
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/containermanagement_protocol.proto?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/containermanagement_protocol.proto (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/containermanagement_protocol.proto Mon Aug 12 21:25:49 2013
@@ -30,7 +30,7 @@ option java_generate_equals_and_hash = t
import "yarn_service_protos.proto";
service ContainerManagementProtocolService {
- rpc startContainer(StartContainerRequestProto) returns (StartContainerResponseProto);
- rpc stopContainer(StopContainerRequestProto) returns (StopContainerResponseProto);
- rpc getContainerStatus(GetContainerStatusRequestProto) returns (GetContainerStatusResponseProto);
+ rpc startContainers(StartContainersRequestProto) returns (StartContainersResponseProto);
+ rpc stopContainers(StopContainersRequestProto) returns (StopContainersResponseProto);
+ rpc getContainerStatuses(GetContainerStatusesRequestProto) returns (GetContainerStatusesResponseProto);
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto Mon Aug 12 21:25:49 2013
@@ -149,6 +149,7 @@ message ApplicationReportProto {
optional ApplicationAttemptIdProto currentApplicationAttemptId = 16;
optional float progress = 17;
optional string applicationType = 18;
+ optional hadoop.common.TokenProto am_rm_token = 19;
}
enum NodeStateProto {
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto Mon Aug 12 21:25:49 2013
@@ -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 RegisterApplicationMasterRespons
}
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 {
@@ -185,3 +182,37 @@ message GetContainerStatusRequestProto {
message GetContainerStatusResponseProto {
optional ContainerStatusProto status = 1;
}
+
+//// bulk API records
+message StartContainersRequestProto {
+ repeated StartContainerRequestProto start_container_request = 1;
+}
+
+message ContainerExceptionMapProto {
+ optional ContainerIdProto container_id = 1;
+ optional SerializedExceptionProto exception = 2;
+}
+
+message StartContainersResponseProto {
+ repeated StringBytesMapProto services_meta_data = 1;
+ repeated ContainerIdProto succeeded_requests = 2;
+ repeated ContainerExceptionMapProto failed_requests = 3;
+}
+
+message StopContainersRequestProto {
+ repeated ContainerIdProto container_id = 1;
+}
+
+message StopContainersResponseProto {
+ repeated ContainerIdProto succeeded_requests = 1;
+ repeated ContainerExceptionMapProto failed_requests = 2;
+}
+
+message GetContainerStatusesRequestProto {
+ repeated ContainerIdProto container_id = 1;
+}
+
+message GetContainerStatusesResponseProto {
+ repeated ContainerStatusProto status = 1;
+ repeated ContainerExceptionMapProto failed_requests = 2;
+}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml Mon Aug 12 21:25:49 2013
@@ -19,12 +19,12 @@
<parent>
<artifactId>hadoop-yarn-applications</artifactId>
<groupId>org.apache.hadoop</groupId>
- <version>2.2.0-SNAPSHOT</version>
+ <version>2.3.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-yarn-applications-distributedshell</artifactId>
- <version>2.2.0-SNAPSHOT</version>
+ <version>2.3.0-SNAPSHOT</version>
<name>hadoop-yarn-applications-distributedshell</name>
<properties>
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java Mon Aug 12 21:25:49 2013
@@ -442,8 +442,8 @@ public class ApplicationMaster {
LOG.info("Starting ApplicationMaster");
AMRMClientAsync.CallbackHandler allocListener = new RMCallbackHandler();
- resourceManager =
- AMRMClientAsync.createAMRMClientAsync(appAttemptID, 1000, allocListener);
+ resourceManager =
+ AMRMClientAsync.createAMRMClientAsync(1000, allocListener);
resourceManager.init(conf);
resourceManager.start();
@@ -483,8 +483,10 @@ public class ApplicationMaster {
// containers
// Keep looping until all the containers are launched and shell script
// executed on them ( regardless of success/failure).
- ContainerRequest containerAsk = setupContainerAskForRM(numTotalContainers);
- resourceManager.addContainerRequest(containerAsk);
+ for (int i = 0; i < numTotalContainers; ++i) {
+ ContainerRequest containerAsk = setupContainerAskForRM();
+ resourceManager.addContainerRequest(containerAsk);
+ }
numRequestedContainers.set(numTotalContainers);
while (!done) {
@@ -591,8 +593,10 @@ public class ApplicationMaster {
numRequestedContainers.addAndGet(askCount);
if (askCount > 0) {
- ContainerRequest containerAsk = setupContainerAskForRM(askCount);
- resourceManager.addContainerRequest(containerAsk);
+ for (int i = 0; i < askCount; ++i) {
+ ContainerRequest containerAsk = setupContainerAskForRM();
+ resourceManager.addContainerRequest(containerAsk);
+ }
}
if (numCompletedContainers.get() == numTotalContainers) {
@@ -645,8 +649,9 @@ public class ApplicationMaster {
}
@Override
- public void onError(Exception e) {
+ public void onError(Throwable e) {
done = true;
+ resourceManager.stop();
}
}
@@ -813,7 +818,7 @@ public class ApplicationMaster {
* @param numContainers Containers to ask for from RM
* @return the setup ResourceRequest to be sent to RM
*/
- private ContainerRequest setupContainerAskForRM(int numContainers) {
+ private ContainerRequest setupContainerAskForRM() {
// setup requirements for hosts
// using * as any host will do for the distributed shell app
// set the priority for the request
@@ -827,7 +832,7 @@ public class ApplicationMaster {
capability.setMemory(containerMemory);
ContainerRequest request = new ContainerRequest(capability, null, null,
- pri, numContainers);
+ pri);
LOG.info("Requested container ask: " + request.toString());
return request;
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/pom.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/pom.xml?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/pom.xml (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/pom.xml Mon Aug 12 21:25:49 2013
@@ -19,12 +19,12 @@
<parent>
<artifactId>hadoop-yarn-applications</artifactId>
<groupId>org.apache.hadoop</groupId>
- <version>2.2.0-SNAPSHOT</version>
+ <version>2.3.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-yarn-applications-unmanaged-am-launcher</artifactId>
- <version>2.2.0-SNAPSHOT</version>
+ <version>2.3.0-SNAPSHOT</version>
<name>hadoop-yarn-applications-unmanaged-am-launcher</name>
<properties>
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/src/main/java/org/apache/hadoop/yarn/applications/unmanagedamlauncher/UnmanagedAMLauncher.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/src/main/java/org/apache/hadoop/yarn/applications/unmanagedamlauncher/UnmanagedAMLauncher.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/src/main/java/org/apache/hadoop/yarn/applications/unmanagedamlauncher/UnmanagedAMLauncher.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/src/main/java/org/apache/hadoop/yarn/applications/unmanagedamlauncher/UnmanagedAMLauncher.java Mon Aug 12 21:25:49 2013
@@ -19,7 +19,9 @@
package org.apache.hadoop.yarn.applications.unmanagedamlauncher;
import java.io.BufferedReader;
+import java.io.DataOutputStream;
import java.io.File;
+import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetAddress;
@@ -36,9 +38,11 @@ import org.apache.commons.cli.ParseExcep
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
-import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
@@ -51,6 +55,7 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
import org.apache.hadoop.yarn.util.Records;
/**
@@ -166,7 +171,34 @@ public class UnmanagedAMLauncher {
return true;
}
- public void launchAM(ApplicationAttemptId attemptId) throws IOException {
+ public void launchAM(ApplicationAttemptId attemptId)
+ throws IOException, YarnException {
+ ApplicationReport report =
+ rmClient.getApplicationReport(attemptId.getApplicationId());
+ if (report.getYarnApplicationState() != YarnApplicationState.ACCEPTED) {
+ throw new YarnException(
+ "Umanaged AM must be in ACCEPTED state before launching");
+ }
+ Credentials credentials = new Credentials();
+ Token<AMRMTokenIdentifier> token =
+ rmClient.getAMRMToken(attemptId.getApplicationId());
+ // Service will be empty but that's okay, we are just passing down only
+ // AMRMToken down to the real AM which eventually sets the correct
+ // service-address.
+ credentials.addToken(token.getService(), token);
+ File tokenFile = File.createTempFile("unmanagedAMRMToken","",
+ new File(System.getProperty("user.dir")));
+ try {
+ FileUtil.chmod(tokenFile.getAbsolutePath(), "600");
+ } catch (InterruptedException ex) {
+ throw new RuntimeException(ex);
+ }
+ tokenFile.deleteOnExit();
+ DataOutputStream os = new DataOutputStream(new FileOutputStream(tokenFile,
+ true));
+ credentials.writeTokenStorageToStream(os);
+ os.close();
+
Map<String, String> env = System.getenv();
ArrayList<String> envAMList = new ArrayList<String>();
boolean setClasspath = false;
@@ -196,6 +228,9 @@ public class UnmanagedAMLauncher {
envAMList.add(ApplicationConstants.APP_SUBMIT_TIME_ENV + "="
+ System.currentTimeMillis());
+ envAMList.add(ApplicationConstants.CONTAINER_TOKEN_FILE_ENV_NAME + "=" +
+ tokenFile.getAbsolutePath());
+
String[] envAM = new String[envAMList.size()];
Process amProc = Runtime.getRuntime().exec(amCmd, envAMList.toArray(envAM));
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/src/test/java/org/apache/hadoop/yarn/applications/unmanagedamlauncher/TestUnmanagedAMLauncher.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/src/test/java/org/apache/hadoop/yarn/applications/unmanagedamlauncher/TestUnmanagedAMLauncher.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/src/test/java/org/apache/hadoop/yarn/applications/unmanagedamlauncher/TestUnmanagedAMLauncher.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/src/test/java/org/apache/hadoop/yarn/applications/unmanagedamlauncher/TestUnmanagedAMLauncher.java Mon Aug 12 21:25:49 2013
@@ -40,7 +40,6 @@ import org.junit.BeforeClass;
import org.junit.Test;
public class TestUnmanagedAMLauncher {
-
private static final Log LOG = LogFactory
.getLog(TestUnmanagedAMLauncher.class);
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/pom.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/pom.xml?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/pom.xml (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/pom.xml Mon Aug 12 21:25:49 2013
@@ -19,12 +19,12 @@
<parent>
<artifactId>hadoop-yarn</artifactId>
<groupId>org.apache.hadoop</groupId>
- <version>2.2.0-SNAPSHOT</version>
+ <version>2.3.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-yarn-applications</artifactId>
- <version>2.2.0-SNAPSHOT</version>
+ <version>2.3.0-SNAPSHOT</version>
<name>hadoop-yarn-applications</name>
<packaging>pom</packaging>
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/pom.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/pom.xml?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/pom.xml (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/pom.xml Mon Aug 12 21:25:49 2013
@@ -17,11 +17,11 @@
<parent>
<artifactId>hadoop-yarn</artifactId>
<groupId>org.apache.hadoop</groupId>
- <version>2.2.0-SNAPSHOT</version>
+ <version>2.3.0-SNAPSHOT</version>
</parent>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-yarn-client</artifactId>
- <version>2.2.0-SNAPSHOT</version>
+ <version>2.3.0-SNAPSHOT</version>
<name>hadoop-yarn-client</name>
<properties>
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java Mon Aug 12 21:25:49 2013
@@ -29,15 +29,15 @@ import org.apache.hadoop.classification.
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 exten
* 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;
}
@@ -69,7 +65,7 @@ public abstract class AMRMClient<T exten
}
/**
- * Object to represent a container request for resources. Scheduler
+ * Object to represent a single container request for resources. Scheduler
* documentation should be consulted for the specifics of how the parameters
* are honored.
*
@@ -101,7 +97,6 @@ public abstract class AMRMClient<T exten
final List<String> nodes;
final List<String> racks;
final Priority priority;
- final int containerCount;
final boolean relaxLocality;
/**
@@ -119,12 +114,10 @@ public abstract class AMRMClient<T exten
* @param priority
* The priority at which to request the containers. Higher
* priorities have lower numerical values.
- * @param containerCount
- * The number of containers to request.
*/
public ContainerRequest(Resource capability, String[] nodes,
- String[] racks, Priority priority, int containerCount) {
- this(capability, nodes, racks, priority, containerCount, true);
+ String[] racks, Priority priority) {
+ this(capability, nodes, racks, priority, true);
}
/**
@@ -141,25 +134,20 @@ public abstract class AMRMClient<T exten
* @param priority
* The priority at which to request the containers. Higher
* priorities have lower numerical values.
- * @param containerCount
- * The number of containers to request.
* @param relaxLocality
* If true, containers for this request may be assigned on hosts
* and racks other than the ones explicitly requested.
*/
public ContainerRequest(Resource capability, String[] nodes,
- String[] racks, Priority priority, int containerCount,
- boolean relaxLocality) {
+ String[] racks, Priority priority, boolean relaxLocality) {
// Validate request
Preconditions.checkArgument(capability != null,
"The Resource to be requested for each container " +
"should not be null ");
Preconditions.checkArgument(priority != null,
"The priority at which to request containers should not be null ");
- Preconditions.checkArgument(containerCount > 0,
- "The number of containers to request should larger than 0");
Preconditions.checkArgument(
- (!relaxLocality && (racks == null || racks.length == 0)
+ !(!relaxLocality && (racks == null || racks.length == 0)
&& (nodes == null || nodes.length == 0)),
"Can't turn off locality relaxation on a " +
"request with no location constraints");
@@ -167,7 +155,6 @@ public abstract class AMRMClient<T exten
this.nodes = (nodes != null ? ImmutableList.copyOf(nodes) : null);
this.racks = (racks != null ? ImmutableList.copyOf(racks) : null);
this.priority = priority;
- this.containerCount = containerCount;
this.relaxLocality = relaxLocality;
}
@@ -187,10 +174,6 @@ public abstract class AMRMClient<T exten
return priority;
}
- public int getContainerCount() {
- return containerCount;
- }
-
public boolean getRelaxLocality() {
return relaxLocality;
}
@@ -199,33 +182,11 @@ public abstract class AMRMClient<T exten
StringBuilder sb = new StringBuilder();
sb.append("Capability[").append(capability).append("]");
sb.append("Priority[").append(priority).append("]");
- sb.append("ContainerCount[").append(containerCount).append("]");
return sb.toString();
}
}
/**
- * This creates a <code>ContainerRequest</code> for 1 container and the
- * AMRMClient stores this request internally. <code>getMatchingRequests</code>
- * can be used to retrieve these requests from AMRMClient. These requests may
- * be matched with an allocated container to determine which request to assign
- * the container to. <code>removeContainerRequest</code> must be called using
- * the same assigned <code>StoredContainerRequest</code> object so that
- * AMRMClient can remove it from its internal store.
- */
- public static class StoredContainerRequest extends ContainerRequest {
- public StoredContainerRequest(Resource capability, String[] nodes,
- String[] racks, Priority priority) {
- super(capability, nodes, racks, priority, 1);
- }
-
- public StoredContainerRequest(Resource capability, String[] nodes,
- String[] racks, Priority priority, boolean relaxLocality) {
- super(capability, nodes, racks, priority, 1, relaxLocality);
- }
- }
-
- /**
* Register the application master. This must be called before any
* other interaction
* @param appHostName Name of the host on which master is running
@@ -311,8 +272,8 @@ public abstract class AMRMClient<T exten
public abstract int getClusterNodeCount();
/**
- * Get outstanding <code>StoredContainerRequest</code>s matching the given
- * parameters. These StoredContainerRequests should have been added via
+ * Get outstanding <code>ContainerRequest</code>s matching the given
+ * parameters. These ContainerRequests should have been added via
* <code>addContainerRequest</code> earlier in the lifecycle. For performance,
* the AMRMClient may return its internal collection directly without creating
* a copy. Users should not perform mutable operations on the return value.
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/YarnClient.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/YarnClient.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/YarnClient.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/YarnClient.java Mon Aug 12 21:25:49 2013
@@ -19,7 +19,6 @@
package org.apache.hadoop.yarn.client.api;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.util.List;
import java.util.Set;
@@ -40,6 +39,7 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.YarnClusterMetrics;
import org.apache.hadoop.yarn.client.api.impl.YarnClientImpl;
import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
@InterfaceAudience.Public
@InterfaceStability.Stable
@@ -54,25 +54,6 @@ public abstract class YarnClient extends
return client;
}
- /**
- * Create a new instance of YarnClient.
- */
- @Public
- public static YarnClient createYarnClient(InetSocketAddress rmAddress) {
- YarnClient client = new YarnClientImpl(rmAddress);
- return client;
- }
-
- /**
- * Create a new instance of YarnClient.
- */
- @Public
- public static YarnClient createYarnClient(String name,
- InetSocketAddress rmAddress) {
- YarnClient client = new YarnClientImpl(name, rmAddress);
- return client;
- }
-
@Private
protected YarnClient(String name) {
super(name);
@@ -161,6 +142,32 @@ public abstract class YarnClient extends
throws YarnException, IOException;
/**
+ * Get the AMRM token of the application.
+ * <p/>
+ * The AMRM token is required for AM to RM scheduling operations. For
+ * managed Application Masters Yarn takes care of injecting it. For unmanaged
+ * Applications Masters, the token must be obtained via this method and set
+ * in the {@link org.apache.hadoop.security.UserGroupInformation} of the
+ * current user.
+ * <p/>
+ * The AMRM token will be returned only if all the following conditions are
+ * met:
+ * <li>
+ * <ul>the requester is the owner of the ApplicationMaster</ul>
+ * <ul>the application master is an unmanaged ApplicationMaster</ul>
+ * <ul>the application master is in ACCEPTED state</ul>
+ * </li>
+ * Else this method returns NULL.
+ *
+ * @param appId {@link ApplicationId} of the application to get the AMRM token
+ * @return the AMRM token if available
+ * @throws YarnException
+ * @throws IOException
+ */
+ public abstract org.apache.hadoop.security.token.Token<AMRMTokenIdentifier>
+ getAMRMToken(ApplicationId appId) throws YarnException, IOException;
+
+ /**
* <p>
* Get a report (ApplicationReport) of all Applications in the cluster.
* </p>
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java Mon Aug 12 21:25:49 2013
@@ -28,7 +28,6 @@ import org.apache.hadoop.classification.
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;
@@ -96,25 +95,19 @@ extends AbstractService {
protected final CallbackHandler handler;
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);
+ public static <T extends ContainerRequest> AMRMClientAsync<T>
+ createAMRMClientAsync(int intervalMs, CallbackHandler callbackHandler) {
+ return new AMRMClientAsyncImpl<T>(intervalMs, callbackHandler);
}
- public static <T extends ContainerRequest> AMRMClientAsync<T>
- createAMRMClientAsync(
- AMRMClient<T> client,
- int intervalMs,
- CallbackHandler callbackHandler) {
+ public static <T extends ContainerRequest> AMRMClientAsync<T>
+ 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
@@ -227,6 +220,13 @@ extends AbstractService {
public float getProgress();
- public void onError(Exception e);
+ /**
+ * Called when error comes from RM communications as well as from errors in
+ * the callback itself from the app. Calling
+ * stop() is the recommended action.
+ *
+ * @param e
+ */
+ public void onError(Throwable e);
}
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java Mon Aug 12 21:25:49 2013
@@ -32,7 +32,6 @@ import org.apache.hadoop.conf.Configurat
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;
@@ -66,11 +65,10 @@ extends AMRMClientAsync<T> {
private volatile boolean keepRunning;
private volatile float progress;
- private volatile Exception savedException;
+ private volatile Throwable 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
@@ -219,23 +217,17 @@ extends AMRMClientAsync<T> {
// synchronization ensures we don't send heartbeats after unregistering
synchronized (unregisterHeartbeatLock) {
if (!keepRunning) {
- break;
+ return;
}
try {
response = client.allocate(progress);
- } catch (YarnException ex) {
- LOG.error("Yarn exception on heartbeat", ex);
+ } catch (Throwable ex) {
+ LOG.error("Exception on heartbeat", ex);
savedException = ex;
// interrupt handler thread in case it waiting on the queue
handlerThread.interrupt();
- break;
- } catch (IOException e) {
- LOG.error("IO exception on heartbeat", e);
- savedException = e;
- // interrupt handler thread in case it waiting on the queue
- handlerThread.interrupt();
- break;
+ return;
}
}
if (response != null) {
@@ -268,51 +260,60 @@ extends AMRMClientAsync<T> {
}
public void run() {
- while (keepRunning) {
- AllocateResponse response;
+ while (true) {
+ if (!keepRunning) {
+ return;
+ }
try {
+ AllocateResponse response;
if(savedException != null) {
LOG.error("Stopping callback due to: ", savedException);
handler.onError(savedException);
- break;
+ return;
+ }
+ try {
+ response = responseQueue.take();
+ } catch (InterruptedException ex) {
+ LOG.info("Interrupted while waiting for queue", ex);
+ continue;
}
- response = responseQueue.take();
- } catch (InterruptedException ex) {
- LOG.info("Interrupted while waiting for queue", ex);
- continue;
- }
- if (response.getAMCommand() != null) {
- switch(response.getAMCommand()) {
- case AM_RESYNC:
- case AM_SHUTDOWN:
- handler.onShutdownRequest();
- LOG.info("Shutdown requested. Stopping callback.");
- return;
- default:
- String msg =
- "Unhandled value of AMCommand: " + response.getAMCommand();
- LOG.error(msg);
- throw new YarnRuntimeException(msg);
+ if (response.getAMCommand() != null) {
+ switch(response.getAMCommand()) {
+ case AM_RESYNC:
+ case AM_SHUTDOWN:
+ handler.onShutdownRequest();
+ LOG.info("Shutdown requested. Stopping callback.");
+ return;
+ default:
+ String msg =
+ "Unhandled value of RM AMCommand: " + response.getAMCommand();
+ LOG.error(msg);
+ throw new YarnRuntimeException(msg);
+ }
+ }
+ List<NodeReport> updatedNodes = response.getUpdatedNodes();
+ if (!updatedNodes.isEmpty()) {
+ handler.onNodesUpdated(updatedNodes);
+ }
+
+ List<ContainerStatus> completed =
+ response.getCompletedContainersStatuses();
+ if (!completed.isEmpty()) {
+ handler.onContainersCompleted(completed);
}
- }
- List<NodeReport> updatedNodes = response.getUpdatedNodes();
- if (!updatedNodes.isEmpty()) {
- handler.onNodesUpdated(updatedNodes);
- }
-
- List<ContainerStatus> completed =
- response.getCompletedContainersStatuses();
- if (!completed.isEmpty()) {
- handler.onContainersCompleted(completed);
- }
- List<Container> allocated = response.getAllocatedContainers();
- if (!allocated.isEmpty()) {
- handler.onContainersAllocated(allocated);
+ List<Container> allocated = response.getAllocatedContainers();
+ if (!allocated.isEmpty()) {
+ handler.onContainersAllocated(allocated);
+ }
+
+ progress = handler.getProgress();
+ } catch (Throwable ex) {
+ handler.onError(ex);
+ // re-throw exception to end the thread
+ throw new YarnRuntimeException(ex);
}
-
- progress = handler.getProgress();
}
}
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java Mon Aug 12 21:25:49 2013
@@ -19,8 +19,6 @@
package org.apache.hadoop.yarn.client.api.impl;
import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -42,30 +40,26 @@ import org.apache.hadoop.classification.
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;
-import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
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;
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.ClientRMProxy;
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.NMTokenCache;
-import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
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.ipc.YarnRPC;
import org.apache.hadoop.yarn.util.RackResolver;
import com.google.common.annotations.VisibleForTesting;
@@ -80,13 +74,9 @@ public class AMRMClientImpl<T extends Co
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;
@@ -157,9 +147,8 @@ public class AMRMClientImpl<T extends Co
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
@@ -171,28 +160,11 @@ public class AMRMClientImpl<T extends Co
@Override
protected void serviceStart() throws Exception {
final YarnConfiguration conf = new YarnConfiguration(getConfig());
- final YarnRPC rpc = YarnRPC.create(conf);
- final InetSocketAddress rmAddress = conf.getSocketAddr(
- YarnConfiguration.RM_SCHEDULER_ADDRESS,
- YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS,
- YarnConfiguration.DEFAULT_RM_SCHEDULER_PORT);
-
- UserGroupInformation currentUser;
try {
- currentUser = UserGroupInformation.getCurrentUser();
+ rmClient = ClientRMProxy.createRMProxy(conf, ApplicationMasterProtocol.class);
} catch (IOException e) {
throw new YarnRuntimeException(e);
}
-
- // CurrentUser should already have AMToken loaded.
- rmClient = currentUser.doAs(new PrivilegedAction<ApplicationMasterProtocol>() {
- @Override
- public ApplicationMasterProtocol run() {
- return (ApplicationMasterProtocol) rpc.getProxy(ApplicationMasterProtocol.class, rmAddress,
- conf);
- }
- });
- LOG.debug("Connecting to ResourceManager at " + rmAddress);
super.serviceStart();
}
@@ -213,18 +185,11 @@ public class AMRMClientImpl<T extends Co
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;
}
@@ -253,8 +218,8 @@ public class AMRMClientImpl<T extends Co
ask.clear();
release.clear();
allocateRequest =
- AllocateRequest.newInstance(appAttemptId, lastResponseId,
- progressIndicator, askList, releaseList, null);
+ AllocateRequest.newInstance(lastResponseId, progressIndicator,
+ askList, releaseList, null);
}
allocateResponse = rmClient.allocate(allocateRequest);
@@ -314,16 +279,9 @@ public class AMRMClientImpl<T extends Co
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);
}
@@ -363,26 +321,26 @@ public class AMRMClientImpl<T extends Co
+ joiner.join(req.getNodes()));
}
for (String node : dedupedNodes) {
- addResourceRequest(req.getPriority(), node, req.getCapability(),
- req.getContainerCount(), req, true);
+ addResourceRequest(req.getPriority(), node, req.getCapability(), req,
+ true);
}
}
for (String rack : dedupedRacks) {
- addResourceRequest(req.getPriority(), rack, req.getCapability(),
- req.getContainerCount(), req, true);
+ addResourceRequest(req.getPriority(), rack, req.getCapability(), req,
+ true);
}
// Ensure node requests are accompanied by requests for
// corresponding rack
for (String rack : inferredRacks) {
- addResourceRequest(req.getPriority(), rack, req.getCapability(),
- req.getContainerCount(), req, req.getRelaxLocality());
+ addResourceRequest(req.getPriority(), rack, req.getCapability(), req,
+ req.getRelaxLocality());
}
// Off-switch
- addResourceRequest(req.getPriority(), ResourceRequest.ANY, req.getCapability(),
- req.getContainerCount(), req, req.getRelaxLocality());
+ addResourceRequest(req.getPriority(), ResourceRequest.ANY,
+ req.getCapability(), req, req.getRelaxLocality());
}
@Override
@@ -398,18 +356,16 @@ public class AMRMClientImpl<T extends Co
// Update resource requests
if (req.getNodes() != null) {
for (String node : new HashSet<String>(req.getNodes())) {
- decResourceRequest(req.getPriority(), node, req.getCapability(),
- req.getContainerCount(), req);
+ decResourceRequest(req.getPriority(), node, req.getCapability(), req);
}
}
for (String rack : allRacks) {
- decResourceRequest(req.getPriority(), rack, req.getCapability(),
- req.getContainerCount(), req);
+ decResourceRequest(req.getPriority(), rack, req.getCapability(), req);
}
- decResourceRequest(req.getPriority(), ResourceRequest.ANY, req.getCapability(),
- req.getContainerCount(), req);
+ decResourceRequest(req.getPriority(), ResourceRequest.ANY,
+ req.getCapability(), req);
}
@Override
@@ -536,7 +492,7 @@ public class AMRMClientImpl<T extends Co
}
private void addResourceRequest(Priority priority, String resourceName,
- Resource capability, int containerCount, T req, boolean relaxLocality) {
+ Resource capability, T req, boolean relaxLocality) {
Map<String, TreeMap<Resource, ResourceRequestInfo>> remoteRequests =
this.remoteRequestsTable.get(priority);
if (remoteRequests == null) {
@@ -564,9 +520,9 @@ public class AMRMClientImpl<T extends Co
}
resourceRequestInfo.remoteRequest.setNumContainers(
- resourceRequestInfo.remoteRequest.getNumContainers() + containerCount);
+ resourceRequestInfo.remoteRequest.getNumContainers() + 1);
- if (req instanceof StoredContainerRequest && relaxLocality) {
+ if (relaxLocality) {
resourceRequestInfo.containerRequests.add(req);
}
@@ -575,7 +531,7 @@ public class AMRMClientImpl<T extends Co
if (LOG.isDebugEnabled()) {
LOG.debug("addResourceRequest:" + " applicationId="
- + appAttemptId + " priority=" + priority.getPriority()
+ + " priority=" + priority.getPriority()
+ " resourceName=" + resourceName + " numContainers="
+ resourceRequestInfo.remoteRequest.getNumContainers()
+ " #asks=" + ask.size());
@@ -585,7 +541,6 @@ public class AMRMClientImpl<T extends Co
private void decResourceRequest(Priority priority,
String resourceName,
Resource capability,
- int containerCount,
T req) {
Map<String, TreeMap<Resource, ResourceRequestInfo>> remoteRequests =
this.remoteRequestsTable.get(priority);
@@ -610,18 +565,16 @@ public class AMRMClientImpl<T extends Co
if (LOG.isDebugEnabled()) {
LOG.debug("BEFORE decResourceRequest:" + " applicationId="
- + appAttemptId + " priority=" + priority.getPriority()
+ + " priority=" + priority.getPriority()
+ " resourceName=" + resourceName + " numContainers="
+ resourceRequestInfo.remoteRequest.getNumContainers()
+ " #asks=" + ask.size());
}
resourceRequestInfo.remoteRequest.setNumContainers(
- resourceRequestInfo.remoteRequest.getNumContainers() - containerCount);
+ resourceRequestInfo.remoteRequest.getNumContainers() - 1);
- if(req instanceof StoredContainerRequest) {
- resourceRequestInfo.containerRequests.remove(req);
- }
+ resourceRequestInfo.containerRequests.remove(req);
if(resourceRequestInfo.remoteRequest.getNumContainers() < 0) {
// guard against spurious removals
@@ -645,7 +598,7 @@ public class AMRMClientImpl<T extends Co
if (LOG.isDebugEnabled()) {
LOG.info("AFTER decResourceRequest:" + " applicationId="
- + appAttemptId + " priority=" + priority.getPriority()
+ + " priority=" + priority.getPriority()
+ " resourceName=" + resourceName + " numContainers="
+ resourceRequestInfo.remoteRequest.getNumContainers()
+ " #asks=" + ask.size());
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/NMClientImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/NMClientImpl.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/NMClientImpl.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/NMClientImpl.java Mon Aug 12 21:25:49 2013
@@ -20,6 +20,8 @@ package org.apache.hadoop.yarn.client.ap
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -30,9 +32,14 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusRequest;
+import org.apache.hadoop.security.token.SecretManager.InvalidToken;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesResponse;
import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.StopContainerRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.StopContainersRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StopContainersResponse;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
@@ -181,11 +188,23 @@ public class NMClientImpl extends NMClie
proxy =
cmProxy.getProxy(container.getNodeId().toString(),
container.getId());
- allServiceResponse =
+ StartContainerRequest scRequest =
+ StartContainerRequest.newInstance(containerLaunchContext,
+ container.getContainerToken());
+ List<StartContainerRequest> list = new ArrayList<StartContainerRequest>();
+ list.add(scRequest);
+ StartContainersRequest allRequests =
+ StartContainersRequest.newInstance(list);
+ StartContainersResponse response =
proxy
- .getContainerManagementProtocol().startContainer(
- StartContainerRequest.newInstance(containerLaunchContext,
- container.getContainerToken())).getAllServicesMetaData();
+ .getContainerManagementProtocol().startContainers(allRequests);
+ if (response.getFailedRequests() != null
+ && response.getFailedRequests().containsKey(container.getId())) {
+ Throwable t =
+ response.getFailedRequests().get(container.getId()).deSerialize();
+ parseAndThrowException(t);
+ }
+ allServiceResponse = response.getAllServicesMetaData();
startingContainer.state = ContainerState.RUNNING;
} catch (YarnException e) {
startingContainer.state = ContainerState.COMPLETE;
@@ -238,11 +257,20 @@ public class NMClientImpl extends NMClie
NodeId nodeId) throws YarnException, IOException {
ContainerManagementProtocolProxyData proxy = null;
+ List<ContainerId> containerIds = new ArrayList<ContainerId>();
+ containerIds.add(containerId);
try {
proxy = cmProxy.getProxy(nodeId.toString(), containerId);
- ContainerStatus containerStatus =
- proxy.getContainerManagementProtocol().getContainerStatus(
- GetContainerStatusRequest.newInstance(containerId)).getStatus();
+ GetContainerStatusesResponse response =
+ proxy.getContainerManagementProtocol().getContainerStatuses(
+ GetContainerStatusesRequest.newInstance(containerIds));
+ if (response.getFailedRequests() != null
+ && response.getFailedRequests().containsKey(containerId)) {
+ Throwable t =
+ response.getFailedRequests().get(containerId).deSerialize();
+ parseAndThrowException(t);
+ }
+ ContainerStatus containerStatus = response.getContainerStatuses().get(0);
return containerStatus;
} finally {
if (proxy != null) {
@@ -254,10 +282,19 @@ public class NMClientImpl extends NMClie
private void stopContainerInternal(ContainerId containerId, NodeId nodeId)
throws IOException, YarnException {
ContainerManagementProtocolProxyData proxy = null;
+ List<ContainerId> containerIds = new ArrayList<ContainerId>();
+ containerIds.add(containerId);
try {
proxy = cmProxy.getProxy(nodeId.toString(), containerId);
- proxy.getContainerManagementProtocol().stopContainer(
- StopContainerRequest.newInstance(containerId));
+ StopContainersResponse response =
+ proxy.getContainerManagementProtocol().stopContainers(
+ StopContainersRequest.newInstance(containerIds));
+ if (response.getFailedRequests() != null
+ && response.getFailedRequests().containsKey(containerId)) {
+ Throwable t = response.getFailedRequests().get(containerId)
+ .deSerialize();
+ parseAndThrowException(t);
+ }
} finally {
if (proxy != null) {
cmProxy.mayBeCloseProxy(proxy);
@@ -285,4 +322,15 @@ public class NMClientImpl extends NMClie
public AtomicBoolean getCleanupRunningContainers() {
return cleanupRunningContainers;
}
+
+ private void parseAndThrowException(Throwable t) throws YarnException,
+ IOException {
+ if (t instanceof YarnException) {
+ throw (YarnException) t;
+ } else if (t instanceof InvalidToken) {
+ throw (InvalidToken) t;
+ } else {
+ throw (IOException) t;
+ }
+ }
}
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/YarnClientImpl.java Mon Aug 12 21:25:49 2013
@@ -33,10 +33,10 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.io.Text;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.yarn.api.ApplicationClientProtocol;
-import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsResponse;
import org.apache.hadoop.yarn.api.protocolrecords.GetClusterNodesRequest;
@@ -59,11 +59,14 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.Token;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.api.records.YarnClusterMetrics;
+import org.apache.hadoop.yarn.client.ClientRMProxy;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.YarnClientApplication;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.apache.hadoop.yarn.ipc.YarnRPC;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
+import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
+import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;
import com.google.common.annotations.VisibleForTesting;
@@ -81,16 +84,7 @@ public class YarnClientImpl extends Yarn
private static final String ROOT = "root";
public YarnClientImpl() {
- this(null);
- }
-
- public YarnClientImpl(InetSocketAddress rmAddress) {
- this(YarnClientImpl.class.getName(), rmAddress);
- }
-
- public YarnClientImpl(String name, InetSocketAddress rmAddress) {
- super(name);
- this.rmAddress = rmAddress;
+ super(YarnClientImpl.class.getName());
}
private static InetSocketAddress getRmAddress(Configuration conf) {
@@ -100,9 +94,7 @@ public class YarnClientImpl extends Yarn
@Override
protected void serviceInit(Configuration conf) throws Exception {
- if (this.rmAddress == null) {
- this.rmAddress = getRmAddress(conf);
- }
+ this.rmAddress = getRmAddress(conf);
statePollIntervalMillis = conf.getLong(
YarnConfiguration.YARN_CLIENT_APP_SUBMISSION_POLL_INTERVAL_MS,
YarnConfiguration.DEFAULT_YARN_CLIENT_APP_SUBMISSION_POLL_INTERVAL_MS);
@@ -111,12 +103,11 @@ public class YarnClientImpl extends Yarn
@Override
protected void serviceStart() throws Exception {
- YarnRPC rpc = YarnRPC.create(getConfig());
-
- this.rmClient = (ApplicationClientProtocol) rpc.getProxy(
- ApplicationClientProtocol.class, rmAddress, getConfig());
- if (LOG.isDebugEnabled()) {
- LOG.debug("Connecting to ResourceManager at " + rmAddress);
+ try {
+ rmClient = ClientRMProxy.createRMProxy(getConfig(),
+ ApplicationClientProtocol.class);
+ } catch (IOException e) {
+ throw new YarnRuntimeException(e);
}
super.serviceStart();
}
@@ -206,6 +197,17 @@ public class YarnClientImpl extends Yarn
return response.getApplicationReport();
}
+ public org.apache.hadoop.security.token.Token<AMRMTokenIdentifier>
+ getAMRMToken(ApplicationId appId) throws YarnException, IOException {
+ Token token = getApplicationReport(appId).getAMRMToken();
+ org.apache.hadoop.security.token.Token<AMRMTokenIdentifier> amrmToken =
+ null;
+ if (token != null) {
+ amrmToken = ConverterUtils.convertFromYarn(token, null);
+ }
+ return amrmToken;
+ }
+
@Override
public List<ApplicationReport> getApplications() throws YarnException,
IOException {
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/NodeCLI.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/NodeCLI.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/NodeCLI.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/NodeCLI.java Mon Aug 12 21:25:49 2013
@@ -143,10 +143,15 @@ public class NodeCLI extends YarnCLI {
nodeReportStr.print("\tContainers : ");
nodeReportStr.println(nodeReport.getNumContainers());
nodeReportStr.print("\tMemory-Used : ");
- nodeReportStr.println((nodeReport.getUsed() == null) ? "0M"
- : (nodeReport.getUsed().getMemory() + "M"));
+ nodeReportStr.println((nodeReport.getUsed() == null) ? "0MB"
+ : (nodeReport.getUsed().getMemory() + "MB"));
nodeReportStr.print("\tMemory-Capacity : ");
- nodeReportStr.println(nodeReport.getCapability().getMemory());
+ nodeReportStr.println(nodeReport.getCapability().getMemory() + "MB");
+ nodeReportStr.print("\tCPU-Used : ");
+ nodeReportStr.println((nodeReport.getUsed() == null) ? "0 vcores"
+ : (nodeReport.getUsed().getVirtualCores() + " vcores"));
+ nodeReportStr.print("\tCPU-Capacity : ");
+ nodeReportStr.println(nodeReport.getCapability().getVirtualCores() + " vcores");
}
if (nodeReport == null) {
Modified: hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/RMAdminCLI.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/RMAdminCLI.java?rev=1513258&r1=1513257&r2=1513258&view=diff
==============================================================================
--- hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/RMAdminCLI.java (original)
+++ hadoop/common/branches/YARN-321/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/RMAdminCLI.java Mon Aug 12 21:25:49 2013
@@ -19,8 +19,6 @@
package org.apache.hadoop.yarn.client.cli;
import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.security.PrivilegedAction;
import java.util.Arrays;
import org.apache.hadoop.classification.InterfaceAudience.Private;
@@ -31,11 +29,11 @@ import org.apache.hadoop.ipc.RemoteExcep
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
+import org.apache.hadoop.yarn.client.ClientRMProxy;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
-import org.apache.hadoop.yarn.ipc.YarnRPC;
import org.apache.hadoop.yarn.server.api.ResourceManagerAdministrationProtocol;
import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshAdminAclsRequest;
import org.apache.hadoop.yarn.server.api.protocolrecords.RefreshNodesRequest;
@@ -164,32 +162,10 @@ public class RMAdminCLI extends Configur
}
}
- private static UserGroupInformation getUGI(Configuration conf
- ) throws IOException {
- return UserGroupInformation.getCurrentUser();
- }
-
private ResourceManagerAdministrationProtocol createAdminProtocol() throws IOException {
// Get the current configuration
final YarnConfiguration conf = new YarnConfiguration(getConf());
-
- // Create the client
- final InetSocketAddress addr = conf.getSocketAddr(
- YarnConfiguration.RM_ADMIN_ADDRESS,
- YarnConfiguration.DEFAULT_RM_ADMIN_ADDRESS,
- YarnConfiguration.DEFAULT_RM_ADMIN_PORT);
- final YarnRPC rpc = YarnRPC.create(conf);
-
- ResourceManagerAdministrationProtocol adminProtocol =
- getUGI(conf).doAs(new PrivilegedAction<ResourceManagerAdministrationProtocol>() {
- @Override
- public ResourceManagerAdministrationProtocol run() {
- return (ResourceManagerAdministrationProtocol) rpc.getProxy(ResourceManagerAdministrationProtocol.class,
- addr, conf);
- }
- });
-
- return adminProtocol;
+ return ClientRMProxy.createRMProxy(conf, ResourceManagerAdministrationProtocol.class);
}
private int refreshQueues() throws IOException, YarnException {